You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/08/25 11:51:50 UTC
(kafka) branch 3.9 updated: KAFKA-17360 local log retention ms/bytes "-2" is not treated correctly (#16932)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new 6d2b81e07f2 KAFKA-17360 local log retention ms/bytes "-2" is not treated correctly (#16932)
6d2b81e07f2 is described below
commit 6d2b81e07f22086e3e752024168772657294c4d9
Author: Kuan-Po Tseng <br...@gmail.com>
AuthorDate: Sun Aug 25 19:46:15 2024 +0800
KAFKA-17360 local log retention ms/bytes "-2" is not treated correctly (#16932)
1) When the local.retention.ms/bytes is set to -2, we didn't replace it with the server-side retention.ms/bytes config, so the -2 local retention won't take effect.
2) When setting retention.ms/bytes to -2, we can notice this log message:
```
Deleting segment LogSegment(baseOffset=10045, size=1037087, lastModifiedTime=1724040653922, largestRecordTimestamp=1724040653835) due to local log retention size -2 breach. Local log size after deletion will be 13435280. (kafka.log.UnifiedLog) [kafka-scheduler-6]
```
This is not helpful for users. We should replace -2 with real retention value when logging.
Reviewers: Luke Chen <sh...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
core/src/main/scala/kafka/log/UnifiedLog.scala | 20 ++++-----
.../test/scala/unit/kafka/log/UnifiedLogTest.scala | 48 ++++++++++++++++++++++
.../kafka/storage/internals/log/LogConfig.java | 14 +++----
.../integration/BaseDeleteSegmentsTest.java | 5 ++-
4 files changed, 68 insertions(+), 19 deletions(-)
diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala
index 69b6787ca63..092b8c92aa4 100644
--- a/core/src/main/scala/kafka/log/UnifiedLog.scala
+++ b/core/src/main/scala/kafka/log/UnifiedLog.scala
@@ -1586,7 +1586,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
startMs - segment.largestTimestamp > retentionMs
}
- deleteOldSegments(shouldDelete, RetentionMsBreach(this, remoteLogEnabled()))
+ deleteOldSegments(shouldDelete, RetentionMsBreach(this, remoteLogEnabledAndRemoteCopyEnabled()))
}
private def deleteRetentionSizeBreachedSegments(): Int = {
@@ -1602,7 +1602,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
}
}
- deleteOldSegments(shouldDelete, RetentionSizeBreach(this, remoteLogEnabled()))
+ deleteOldSegments(shouldDelete, RetentionSizeBreach(this, remoteLogEnabledAndRemoteCopyEnabled()))
}
private def deleteLogStartOffsetBreachedSegments(): Int = {
@@ -2347,11 +2347,11 @@ object UnifiedLog extends Logging {
}
private[log] def localRetentionMs(config: LogConfig, remoteLogEnabledAndRemoteCopyEnabled: Boolean): Long = {
- if (remoteLogEnabledAndRemoteCopyEnabled) config.remoteLogConfig.localRetentionMs else config.retentionMs
+ if (remoteLogEnabledAndRemoteCopyEnabled) config.localRetentionMs else config.retentionMs
}
private[log] def localRetentionSize(config: LogConfig, remoteLogEnabledAndRemoteCopyEnabled: Boolean): Long = {
- if (remoteLogEnabledAndRemoteCopyEnabled) config.remoteLogConfig.localRetentionBytes else config.retentionSize
+ if (remoteLogEnabledAndRemoteCopyEnabled) config.localRetentionBytes else config.retentionSize
}
}
@@ -2367,19 +2367,19 @@ object LogMetricNames {
}
}
-case class RetentionMsBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends SegmentDeletionReason {
+case class RetentionMsBreach(log: UnifiedLog, remoteLogEnabledAndRemoteCopyEnabled: Boolean) extends SegmentDeletionReason {
override def logReason(toDelete: List[LogSegment]): Unit = {
- val retentionMs = UnifiedLog.localRetentionMs(log.config, remoteLogEnabled)
+ val retentionMs = UnifiedLog.localRetentionMs(log.config, remoteLogEnabledAndRemoteCopyEnabled)
toDelete.foreach { segment =>
if (segment.largestRecordTimestamp.isPresent)
- if (remoteLogEnabled)
+ if (remoteLogEnabledAndRemoteCopyEnabled)
log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the largest " +
s"record timestamp in the segment")
else
log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the largest " +
s"record timestamp in the segment")
else {
- if (remoteLogEnabled)
+ if (remoteLogEnabledAndRemoteCopyEnabled)
log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the " +
s"last modified time of the segment")
else
@@ -2390,12 +2390,12 @@ case class RetentionMsBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends
}
}
-case class RetentionSizeBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends SegmentDeletionReason {
+case class RetentionSizeBreach(log: UnifiedLog, remoteLogEnabledAndRemoteCopyEnabled: Boolean) extends SegmentDeletionReason {
override def logReason(toDelete: List[LogSegment]): Unit = {
var size = log.size
toDelete.foreach { segment =>
size -= segment.size
- if (remoteLogEnabled) log.info(s"Deleting segment $segment due to local log retention size ${UnifiedLog.localRetentionSize(log.config, remoteLogEnabled)} breach. " +
+ if (remoteLogEnabledAndRemoteCopyEnabled) log.info(s"Deleting segment $segment due to local log retention size ${UnifiedLog.localRetentionSize(log.config, remoteLogEnabledAndRemoteCopyEnabled)} breach. " +
s"Local log size after deletion will be $size.")
else log.info(s"Deleting segment $segment due to log retention size ${log.config.retentionSize} breach. Log size " +
s"after deletion will be $size.")
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index b45c9171d66..6d7614d9b5c 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -4187,6 +4187,54 @@ class UnifiedLogTest {
assertEquals(1, log.logSegments.size)
}
+ @Test
+ def testRetentionOnLocalLogDeletionWhenRemoteLogCopyEnabledAndDefaultLocalRetentionBytes(): Unit = {
+ def createRecords = TestUtils.records(List(new SimpleRecord(mockTime.milliseconds(), "a".getBytes)))
+ val segmentBytes = createRecords.sizeInBytes()
+ val retentionBytesConfig = LogTestUtils.createLogConfig(segmentBytes = segmentBytes, retentionBytes = 1,
+ fileDeleteDelayMs = 0, remoteLogStorageEnable = true)
+ val log = createLog(logDir, retentionBytesConfig, remoteStorageSystemEnable = true)
+
+ // Given 6 segments of 1 message each
+ for (_ <- 0 until 6) {
+ log.appendAsLeader(createRecords, leaderEpoch = 0)
+ }
+ assertEquals(6, log.logSegments.size)
+
+ log.updateHighWatermark(log.logEndOffset)
+ // simulate calls to upload 2 segments to remote storage
+ log.updateHighestOffsetInRemoteStorage(1)
+ log.deleteOldSegments()
+ assertEquals(4, log.logSegments.size())
+ assertEquals(0, log.logStartOffset)
+ assertEquals(2, log.localLogStartOffset())
+ }
+
+ @Test
+ def testRetentionOnLocalLogDeletionWhenRemoteLogCopyEnabledAndDefaultLocalRetentionMs(): Unit = {
+ def createRecords = TestUtils.records(List(new SimpleRecord(mockTime.milliseconds(), "a".getBytes)))
+ val segmentBytes = createRecords.sizeInBytes()
+ val retentionBytesConfig = LogTestUtils.createLogConfig(segmentBytes = segmentBytes, retentionMs = 1000,
+ fileDeleteDelayMs = 0, remoteLogStorageEnable = true)
+ val log = createLog(logDir, retentionBytesConfig, remoteStorageSystemEnable = true)
+
+ // Given 6 segments of 1 message each
+ for (_ <- 0 until 6) {
+ log.appendAsLeader(createRecords, leaderEpoch = 0)
+ }
+ assertEquals(6, log.logSegments.size)
+
+ log.updateHighWatermark(log.logEndOffset)
+ // simulate calls to upload 2 segments to remote storage
+ log.updateHighestOffsetInRemoteStorage(1)
+
+ mockTime.sleep(1001)
+ log.deleteOldSegments()
+ assertEquals(4, log.logSegments.size())
+ assertEquals(0, log.logStartOffset)
+ assertEquals(2, log.localLogStartOffset())
+ }
+
@Test
def testRetentionOnLocalLogDeletionWhenRemoteLogCopyDisabled(): Unit = {
def createRecords = TestUtils.records(List(new SimpleRecord(mockTime.milliseconds(), "a".getBytes)))
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
index f3f151ddc18..3299b6e89d7 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
@@ -108,13 +108,13 @@ public class LogConfig extends AbstractConfig {
}
}
- public static class RemoteLogConfig {
+ private static class RemoteLogConfig {
- public final boolean remoteStorageEnable;
- public final boolean remoteLogDeleteOnDisable;
- public final boolean remoteLogCopyDisable;
- public final long localRetentionMs;
- public final long localRetentionBytes;
+ private final boolean remoteStorageEnable;
+ private final boolean remoteLogDeleteOnDisable;
+ private final boolean remoteLogCopyDisable;
+ private final long localRetentionMs;
+ private final long localRetentionBytes;
private RemoteLogConfig(LogConfig config) {
this.remoteStorageEnable = config.getBoolean(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG);
@@ -375,8 +375,8 @@ public class LogConfig extends AbstractConfig {
public final List<String> leaderReplicationThrottledReplicas;
public final List<String> followerReplicationThrottledReplicas;
public final boolean messageDownConversionEnable;
- public final RemoteLogConfig remoteLogConfig;
+ private final RemoteLogConfig remoteLogConfig;
private final int maxMessageSize;
private final Map<?, ?> props;
diff --git a/storage/src/test/java/org/apache/kafka/tiered/storage/integration/BaseDeleteSegmentsTest.java b/storage/src/test/java/org/apache/kafka/tiered/storage/integration/BaseDeleteSegmentsTest.java
index 0147c9f7c54..11c3ca7ad09 100644
--- a/storage/src/test/java/org/apache/kafka/tiered/storage/integration/BaseDeleteSegmentsTest.java
+++ b/storage/src/test/java/org/apache/kafka/tiered/storage/integration/BaseDeleteSegmentsTest.java
@@ -23,6 +23,7 @@ import org.apache.kafka.tiered.storage.specs.KeyValueSpec;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import static org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.DELETE_SEGMENT;
@@ -54,8 +55,8 @@ public abstract class BaseDeleteSegmentsTest extends TieredStorageTestHarness {
.expectSegmentToBeOffloaded(broker0, topicA, p0, 1, new KeyValueSpec("k1", "v1"))
.expectSegmentToBeOffloaded(broker0, topicA, p0, 2, new KeyValueSpec("k2", "v2"))
.expectEarliestLocalOffsetInLogDirectory(topicA, p0, 3L)
- .produce(topicA, p0, new KeyValueSpec("k0", "v0"), new KeyValueSpec("k1", "v1"),
- new KeyValueSpec("k2", "v2"), new KeyValueSpec("k3", "v3"))
+ .produceWithTimestamp(topicA, p0, new KeyValueSpec("k0", "v0"), new KeyValueSpec("k1", "v1"),
+ new KeyValueSpec("k2", "v2"), new KeyValueSpec("k3", "v3", System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1)))
// update the topic config such that it triggers the deletion of segments
.updateTopicConfig(topicA, configsToBeAdded(), Collections.emptyList())
// expect that the three offloaded remote log segments are deleted