This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new 52f6edacf1b KAFKA-20732: seed highestOffsetInRemoteStorage before
size-based remote retention (#22661)
52f6edacf1b is described below
commit 52f6edacf1b15c29f5285781df779393b7ad6b0a
Author: Adam Horky <[email protected]>
AuthorDate: Wed Jul 1 07:58:15 2026 +0200
KAFKA-20732: seed highestOffsetInRemoteStorage before size-based remote
retention (#22661)
JIRA: https://issues.apache.org/jira/browse/KAFKA-20732
**Problem**
On a freshly elected leader of a tiered partition, `RemoteLogManager`
size-retention can run while `highestOffsetInRemoteStorage` is still
`-1` — it is seeded only by the copy and follower tasks, which race the
expiration task (both are scheduled with `initialDelay=0` on separate
thread pools). `UnifiedLog.onlyLocalLogSegmentsSize()` then filters
`baseOffset >= -1` and counts the entire local log, which is also
counted in `remoteLogSizeBytes` for the same offsets. The apparent size
roughly doubles, a false `retention.bytes` breach fires, and
in-retention data is deleted from the remote tier and — via the
resulting `logStartOffset` advance, replicated to followers — the local
tier on all replicas.
The follower path already guards this: `RLMFollowerTask.execute()` seeds
the offset with the comment *"so that the local log segments are not
deleted before they are copied to remote storage."* The
leader/expiration path, which runs size-retention, had no equivalent
step.
**Fix**
In `cleanupExpiredRemoteLogSegments`, when
`highestOffsetInRemoteStorage() == -1`, seed it from remote metadata
(`findHighestRemoteOffset`) before computing size-retention. It is a
no-op once the offset is already seeded.
**Testing**
Added
`RemoteLogManagerTest.testSizeRetentionDoesNotOverDeleteOnFreshLeaderUntilHighestRemoteOffsetSeeded`
— fails without the fix (deletes both copied segments), passes with it.
Also reproduced and fix-validated on a live 3.9.0 cluster (details in
the JIRA).
Reviewers: Luke Chen <[email protected]>, Kamal Chandraprakash
<[email protected]>
---
.../log/remote/storage/RemoteLogManager.java | 18 +++++++
.../log/remote/storage/RemoteLogManagerTest.java | 59 ++++++++++++++++++++++
2 files changed, 77 insertions(+)
diff --git
a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
index 27713827fef..8f521f623e4 100644
---
a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
+++
b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
@@ -1347,6 +1347,20 @@ public class RemoteLogManager implements Closeable,
AsyncOffsetReader {
return new RemoteLogMetadataStats(epochsSet, metadataCount,
sizeInBytes, copyFinishedSegmentsSizeInBytes);
}
+ // On a freshly-elected leader, highestOffsetInRemoteStorage is
unseeded (-1) until RLMCopyTask /
+ // RLMFollowerTask seed it; RLMExpirationTask does not. If size-based
retention runs while it is -1,
+ // UnifiedLog.onlyLocalLogSegmentsSize() (filter baseOffset >=
highestOffsetInRemoteStorage()) returns
+ // the whole local log -- including segments already copied to remote
-- which buildRetentionSizeData
+ // then double-counts against the remote size, over-deleting
in-retention data from both tiers.
+ private void maybeSeedHighestOffsetInRemoteStorage(UnifiedLog log)
throws RemoteStorageException {
+ if (log.highestOffsetInRemoteStorage() == -1) {
+ OffsetAndEpoch highestRemoteOffsetAndEpoch =
findHighestRemoteOffset(topicIdPartition, log);
+ if (highestRemoteOffsetAndEpoch.offset() >= 0) {
+
log.updateHighestOffsetInRemoteStorage(highestRemoteOffsetAndEpoch.offset());
+ }
+ }
+ }
+
/**
* Cleanup expired and dangling remote log segments.
*/
@@ -1383,6 +1397,10 @@ public class RemoteLogManager implements Closeable,
AsyncOffsetReader {
// Build the leader epoch map by filtering the epochs that do not
have any records.
NavigableMap<Integer, Long> epochWithOffsets =
buildFilteredLeaderEpochMap(leaderEpochCache.epochWithOffsets());
+ // Seed highestOffsetInRemoteStorage before size-retention to
avoid the fresh-leader double-count
+ // (no-op once RLMCopyTask/RLMFollowerTask have seeded it).
+ maybeSeedHighestOffsetInRemoteStorage(log);
+
long logStartOffset = log.logStartOffset();
long logEndOffset = log.logEndOffset();
Optional<RetentionSizeData> retentionSizeData =
buildRetentionSizeData(log.config().retentionSize,
diff --git
a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
index 234bb0f06f4..b05bb754eed 100644
---
a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
+++
b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
@@ -2809,6 +2809,65 @@ public class RemoteLogManagerTest {
}
}
+ @Test
+ public void
testSizeRetentionDoesNotOverDeleteOnFreshLeaderUntilHighestRemoteOffsetSeeded()
+ throws RemoteStorageException, ExecutionException,
InterruptedException {
+ // Regression test: on a freshly-elected leader,
highestOffsetInRemoteStorage is unseeded (-1) until
+ // RLMCopyTask/RLMFollowerTask seed it. While it is -1, the real
UnifiedLog.onlyLocalLogSegmentsSize()
+ // (filter baseOffset >= highestOffsetInRemoteStorage()) returns the
whole local log -- including
+ // segments already copied to remote -- so buildRetentionSizeData
double-counts that overlap against the
+ // remote size and over-deletes both tiers. Here
onlyLocalLogSegmentsSize() is wired to the real, mutable
+ // highestOffsetInRemoteStorage (the field the fix seeds) with that
exact behaviour:
+ // unseeded (-1) -> whole local log (2048, the
copied-but-not-yet-evicted overlap)
+ // seeded -> only the uncopied tail (0)
+ // RED without the seeding fix: highest stays -1 ->
onlyLocalLogSegmentsSize()=2048 ->
+ // total = 2048 (remote) + 2048 = 4096 > 2048 -> both segments
deleted, logStartOffset -> 200.
+ // GREEN with it: RLMExpirationTask seeds highest=199 ->
onlyLocalLogSegmentsSize()=0 ->
+ // total = 2048 == cap -> nothing deleted.
+ long segmentSize = 1024L;
+ long retentionSize = 2 * segmentSize;
+ AtomicLong highestRemote = new AtomicLong(-1L); // fresh leader:
unseeded
+ when(mockLog.highestOffsetInRemoteStorage()).thenAnswer(inv ->
highestRemote.get());
+ doAnswer(inv -> {
+ long offset = inv.getArgument(0);
+ if (offset > highestRemote.get()) {
+ highestRemote.set(offset);
+ }
+ return null;
+ }).when(mockLog).updateHighestOffsetInRemoteStorage(anyLong());
+ when(mockLog.onlyLocalLogSegmentsSize()).thenAnswer(inv ->
highestRemote.get() < 0 ? 2 * segmentSize : 0L);
+
+ Map<String, Long> logProps = new HashMap<>();
+ logProps.put("retention.bytes", retentionSize);
+ logProps.put("retention.ms", -1L);
+ when(mockLog.config()).thenReturn(new LogConfig(logProps));
+
+ List<EpochEntry> epochEntries = List.of(epochEntry0);
+ checkpoint.write(epochEntries);
+ LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint,
scheduler);
+ when(mockLog.leaderEpochCache()).thenReturn(cache);
+
when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition());
+ when(mockLog.logEndOffset()).thenReturn(200L);
+ when(mockLog.size()).thenReturn(2 * segmentSize);
+
+ // 2 copied remote segments [0..99],[100..199], 1024 each; the seeding
fix derives highest from these.
+ List<RemoteLogSegmentMetadata> metadataList =
listRemoteLogSegmentMetadata(
+ leaderTopicIdPartition, 2, 100, (int) segmentSize,
epochEntries, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+
when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition))
+ .thenAnswer(ans -> metadataList.iterator());
+
when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, 0))
+ .thenAnswer(ans -> metadataList.iterator());
+
when(remoteLogMetadataManager.updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class)))
+ .thenReturn(CompletableFuture.runAsync(() -> { }));
+
when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class),
anyInt()))
+ .thenReturn(Optional.of(199L));
+
+ remoteLogManager.new
RLMExpirationTask(leaderTopicIdPartition).cleanupExpiredRemoteLogSegments();
+
+ verify(remoteStorageManager, never()).deleteLogSegmentData(any());
+ assertEquals(0L, currentLogStartOffset.get());
+ }
+
@ParameterizedTest(name =
"testDeletionOnOverlappingRetentionBreachedSegments retentionSize={0}
retentionMs={1}")
@CsvSource(value = {"0, -1", "-1, 0"})
public void testDeletionOnOverlappingRetentionBreachedSegments(long
retentionSize,
