kamalcph commented on code in PR #23246:
URL: https://github.com/apache/kafka/pull/23246#discussion_r3863784463
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java:
##########
@@ -467,6 +467,84 @@ public void
shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention() thr
assertTrue(deletedSegments > 0, "At least one segment should be
deleted");
}
+ @Test
+ public void onlyLocalLogSegmentsExcludeCopiedSingleRecordBoundarySegment()
throws IOException {
+ // Regression test: onlyLocalLogSegments{Count,Size} must exclude a
closed segment whose
+ // base-offset equals highestOffsetInRemoteStorage. That segment is
the last one already
+ // copied to remote (highestOffsetInRemoteStorage holds the *end*
offset of the last copied
+ // segment). For single-record segments (base-offset == end-offset,
common on low-throughput
+ // partitions) a ">=" comparison wrongly counted the already-copied
boundary segment as local.
+ // Since RemoteLogManager derives RemoteCopyLagSegments as
(onlyLocalLogSegmentsCount - 1) and
+ // only refreshes it after a copy, that produced a permanent phantom
lag of 1 on idle partitions.
Review Comment:
Shall we remove this comment? It seems redundant. The java-doc in the source
code already explains it and the test asserts it.
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java:
##########
@@ -467,6 +467,84 @@ public void
shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention() thr
assertTrue(deletedSegments > 0, "At least one segment should be
deleted");
}
+ @Test
+ public void onlyLocalLogSegmentsExcludeCopiedSingleRecordBoundarySegment()
throws IOException {
+ // Regression test: onlyLocalLogSegments{Count,Size} must exclude a
closed segment whose
+ // base-offset equals highestOffsetInRemoteStorage. That segment is
the last one already
+ // copied to remote (highestOffsetInRemoteStorage holds the *end*
offset of the last copied
+ // segment). For single-record segments (base-offset == end-offset,
common on low-throughput
+ // partitions) a ">=" comparison wrongly counted the already-copied
boundary segment as local.
+ // Since RemoteLogManager derives RemoteCopyLagSegments as
(onlyLocalLogSegmentsCount - 1) and
+ // only refreshes it after a copy, that produced a permanent phantom
lag of 1 on idle partitions.
+ 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());
+ assertEquals(1L, log.onlyLocalLogSegmentsCount() - 1,
Review Comment:
can we remove this assert?
```suggestion
```
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java:
##########
@@ -467,6 +467,84 @@ public void
shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention() thr
assertTrue(deletedSegments > 0, "At least one segment should be
deleted");
}
+ @Test
+ public void onlyLocalLogSegmentsExcludeCopiedSingleRecordBoundarySegment()
throws IOException {
Review Comment:
shall we merge both the unit tests? A partition can contain segments with 1
or multiple records:
(eg)
```java
@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());
}
```
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java:
##########
@@ -467,6 +467,84 @@ public void
shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention() thr
assertTrue(deletedSegments > 0, "At least one segment should be
deleted");
}
+ @Test
+ public void onlyLocalLogSegmentsExcludeCopiedSingleRecordBoundarySegment()
throws IOException {
+ // Regression test: onlyLocalLogSegments{Count,Size} must exclude a
closed segment whose
+ // base-offset equals highestOffsetInRemoteStorage. That segment is
the last one already
+ // copied to remote (highestOffsetInRemoteStorage holds the *end*
offset of the last copied
+ // segment). For single-record segments (base-offset == end-offset,
common on low-throughput
+ // partitions) a ">=" comparison wrongly counted the already-copied
boundary segment as local.
+ // Since RemoteLogManager derives RemoteCopyLagSegments as
(onlyLocalLogSegmentsCount - 1) and
+ // only refreshes it after a copy, that produced a permanent phantom
lag of 1 on idle partitions.
+ 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());
+ assertEquals(1L, log.onlyLocalLogSegmentsCount() - 1,
+ "RemoteCopyLagSegments should be 1 (one uncopied closed
segment)");
+
+ // 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(0L, log.onlyLocalLogSegmentsCount() - 1,
Review Comment:
can we remove this assert? The previous line already asserts it:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
