Skip to content

Commit 25f376f

Browse files
unknowntpounknowntpo
committed
MINOR: Use named retention sentinels
1 parent f5e387e commit 25f376f

2 files changed

Lines changed: 22 additions & 21 deletions

File tree

‎server/src/main/java/org/apache/kafka/server/config/DynamicLogConfig.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -79,10 +79,10 @@ private AbstractKafkaConfig requireKafkaConfig(AbstractConfig config) {
7979
private void validateLogLocalRetentionMs(AbstractKafkaConfig config) {
8080
long logRetentionMs = config.logRetentionTimeMillis();
8181
long logLocalRetentionMs = config.getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP);
82-
if (logRetentionMs != -1L && logLocalRetentionMs != -2L) {
83-
if (logLocalRetentionMs == -1L) {
82+
if (logRetentionMs != LogConfig.NO_RETENTION_LIMIT && logLocalRetentionMs != LogConfig.DEFAULT_LOCAL_RETENTION_MS) {
83+
if (logLocalRetentionMs == LogConfig.NO_RETENTION_LIMIT) {
8484
throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs,
85-
"Value must not be -1 as " + ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG + " value is set as " + logRetentionMs + ".");
85+
"Value must not be " + LogConfig.NO_RETENTION_LIMIT + " as " + ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG + " value is set as " + logRetentionMs + ".");
8686
}
8787
if (logLocalRetentionMs > logRetentionMs) {
8888
throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs,
@@ -94,10 +94,10 @@ private void validateLogLocalRetentionMs(AbstractKafkaConfig config) {
9494
private void validateLogLocalRetentionBytes(AbstractKafkaConfig config) {
9595
long logRetentionBytes = config.logRetentionBytes();
9696
long logLocalRetentionBytes = config.getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP);
97-
if (logRetentionBytes > -1L && logLocalRetentionBytes != -2L) {
98-
if (logLocalRetentionBytes == -1L) {
97+
if (logRetentionBytes > LogConfig.NO_RETENTION_LIMIT && logLocalRetentionBytes != LogConfig.DEFAULT_LOCAL_RETENTION_BYTES) {
98+
if (logLocalRetentionBytes == LogConfig.NO_RETENTION_LIMIT) {
9999
throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes,
100-
"Value must not be -1 as " + ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG + " value is set as " + logRetentionBytes + ".");
100+
"Value must not be " + LogConfig.NO_RETENTION_LIMIT + " as " + ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG + " value is set as " + logRetentionBytes + ".");
101101
}
102102
if (logLocalRetentionBytes > logRetentionBytes) {
103103
throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes,
@@ -109,7 +109,7 @@ private void validateLogLocalRetentionBytes(AbstractKafkaConfig config) {
109109
private void validateLogRemoteCopyLagMs(AbstractKafkaConfig config) {
110110
long logRetentionMs = config.logRetentionTimeMillis();
111111
long logLocalRetentionMs = config.getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP);
112-
long effectiveLocalRetentionMs = logLocalRetentionMs == -2L ? logRetentionMs : logLocalRetentionMs;
112+
long effectiveLocalRetentionMs = logLocalRetentionMs == LogConfig.DEFAULT_LOCAL_RETENTION_MS ? logRetentionMs : logLocalRetentionMs;
113113
long logRemoteCopyLagMs = config.getLong(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_MS_PROP);
114114
if (logRemoteCopyLagMs > 0L && effectiveLocalRetentionMs >= 0L && logRemoteCopyLagMs > effectiveLocalRetentionMs) {
115115
throw new ConfigException(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_MS_PROP, logRemoteCopyLagMs,
@@ -121,7 +121,7 @@ private void validateLogRemoteCopyLagMs(AbstractKafkaConfig config) {
121121
private void validateLogRemoteCopyLagBytes(AbstractKafkaConfig config) {
122122
long logRetentionBytes = config.logRetentionBytes();
123123
long logLocalRetentionBytes = config.getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP);
124-
long effectiveLocalRetentionBytes = logLocalRetentionBytes == -2L ? logRetentionBytes : logLocalRetentionBytes;
124+
long effectiveLocalRetentionBytes = logLocalRetentionBytes == LogConfig.DEFAULT_LOCAL_RETENTION_BYTES ? logRetentionBytes : logLocalRetentionBytes;
125125
long logRemoteCopyLagBytes = config.getLong(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_BYTES_PROP);
126126
if (logRemoteCopyLagBytes > 0L && effectiveLocalRetentionBytes >= 0L && logRemoteCopyLagBytes > effectiveLocalRetentionBytes) {
127127
throw new ConfigException(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_BYTES_PROP, logRemoteCopyLagBytes,

‎storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java‎

Lines changed: 14 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,7 @@ public Optional<String> serverConfigName(String configName) {
142142
public static final boolean DEFAULT_REMOTE_STORAGE_ENABLE = false;
143143
public static final boolean DEFAULT_REMOTE_LOG_COPY_DISABLE_CONFIG = false;
144144
public static final boolean DEFAULT_REMOTE_LOG_DELETE_ON_DISABLE_CONFIG = false;
145+
public static final long NO_RETENTION_LIMIT = -1L; // It indicates no retention limit
145146
public static final long DEFAULT_LOCAL_RETENTION_BYTES = -2; // It indicates the value to be derived from RetentionBytes
146147
public static final long DEFAULT_LOCAL_RETENTION_MS = -2; // It indicates the value to be derived from RetentionMs
147148
public static final long DEFAULT_REMOTE_COPY_LAG_MS = 0;
@@ -209,7 +210,7 @@ public Optional<String> serverConfigName(String configName) {
209210
// can be negative. See kafka.log.LogManager.cleanupSegmentsToMaintainSize
210211
.define(TopicConfig.RETENTION_BYTES_CONFIG, LONG, ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, MEDIUM, TopicConfig.RETENTION_BYTES_DOC)
211212
// can be negative. See kafka.log.LogManager.cleanupExpiredSegments
212-
.define(TopicConfig.RETENTION_MS_CONFIG, LONG, DEFAULT_RETENTION_MS, atLeast(-1), MEDIUM,
213+
.define(TopicConfig.RETENTION_MS_CONFIG, LONG, DEFAULT_RETENTION_MS, atLeast(NO_RETENTION_LIMIT), MEDIUM,
213214
TopicConfig.RETENTION_MS_DOC)
214215
.define(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, INT, ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT, atLeast(0), MEDIUM,
215216
TopicConfig.MAX_MESSAGE_BYTES_DOC)
@@ -257,8 +258,8 @@ public Optional<String> serverConfigName(String configName) {
257258
.define(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, LONG, DEFAULT_LOCAL_RETENTION_BYTES, atLeast(-2), MEDIUM,
258259
TopicConfig.LOCAL_LOG_RETENTION_BYTES_DOC)
259260
.define(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, BOOLEAN, false, MEDIUM, TopicConfig.REMOTE_LOG_COPY_DISABLE_DOC)
260-
.define(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG, LONG, DEFAULT_REMOTE_COPY_LAG_MS, atLeast(-1), MEDIUM, TopicConfig.REMOTE_COPY_LAG_MS_DOC)
261-
.define(TopicConfig.REMOTE_COPY_LAG_BYTES_CONFIG, LONG, DEFAULT_REMOTE_COPY_LAG_BYTES, atLeast(-1), MEDIUM, TopicConfig.REMOTE_COPY_LAG_BYTES_DOC)
261+
.define(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG, LONG, DEFAULT_REMOTE_COPY_LAG_MS, atLeast(MAX_REMOTE_COPY_LAG_MS), MEDIUM, TopicConfig.REMOTE_COPY_LAG_MS_DOC)
262+
.define(TopicConfig.REMOTE_COPY_LAG_BYTES_CONFIG, LONG, DEFAULT_REMOTE_COPY_LAG_BYTES, atLeast(MAX_REMOTE_COPY_LAG_BYTES), MEDIUM, TopicConfig.REMOTE_COPY_LAG_BYTES_DOC)
262263
.define(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, BOOLEAN, false, MEDIUM, TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_DOC)
263264
.define(TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, BOOLEAN, false, MEDIUM, TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_DOC)
264265
.defineInternal(INTERNAL_SEGMENT_BYTES_CONFIG, INT, null, null, MEDIUM, INTERNAL_SEGMENT_BYTES_DOC);
@@ -600,10 +601,10 @@ private static void validateRemoteStorageRequiresDeleteCleanupPolicy(Map<String,
600601
private static void validateRemoteStorageRetentionSize(Map<String, ?> props) {
601602
Long retentionBytes = (Long) props.get(TopicConfig.RETENTION_BYTES_CONFIG);
602603
Long localRetentionBytes = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG);
603-
if (retentionBytes > -1 && localRetentionBytes != -2) {
604-
if (localRetentionBytes == -1) {
605-
String message = String.format("Value must not be -1 as %s value is set as %d.",
606-
TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes);
604+
if (retentionBytes > NO_RETENTION_LIMIT && localRetentionBytes != DEFAULT_LOCAL_RETENTION_BYTES) {
605+
if (localRetentionBytes == NO_RETENTION_LIMIT) {
606+
String message = String.format("Value must not be %d as %s value is set as %d.",
607+
NO_RETENTION_LIMIT, TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes);
607608
throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes, message);
608609
}
609610
if (localRetentionBytes > retentionBytes) {
@@ -617,10 +618,10 @@ private static void validateRemoteStorageRetentionSize(Map<String, ?> props) {
617618
private static void validateRemoteStorageRetentionTime(Map<String, ?> props) {
618619
Long retentionMs = (Long) props.get(TopicConfig.RETENTION_MS_CONFIG);
619620
Long localRetentionMs = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG);
620-
if (retentionMs != -1 && localRetentionMs != -2) {
621-
if (localRetentionMs == -1) {
622-
String message = String.format("Value must not be -1 as %s value is set as %d.",
623-
TopicConfig.RETENTION_MS_CONFIG, retentionMs);
621+
if (retentionMs != NO_RETENTION_LIMIT && localRetentionMs != DEFAULT_LOCAL_RETENTION_MS) {
622+
if (localRetentionMs == NO_RETENTION_LIMIT) {
623+
String message = String.format("Value must not be %d as %s value is set as %d.",
624+
NO_RETENTION_LIMIT, TopicConfig.RETENTION_MS_CONFIG, retentionMs);
624625
throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs, message);
625626
}
626627
if (localRetentionMs > retentionMs) {
@@ -635,7 +636,7 @@ private static void validateRemoteCopyLagTime(Map<?, ?> props) {
635636
Long retentionMs = (Long) props.get(TopicConfig.RETENTION_MS_CONFIG);
636637
Long localRetentionMs = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG);
637638
Long remoteCopyLagMs = (Long) props.get(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG);
638-
long effectiveLocalRetentionMs = localRetentionMs == -2 ? retentionMs : localRetentionMs;
639+
long effectiveLocalRetentionMs = localRetentionMs == DEFAULT_LOCAL_RETENTION_MS ? retentionMs : localRetentionMs;
639640
if (remoteCopyLagMs > 0 && effectiveLocalRetentionMs >= 0
640641
&& remoteCopyLagMs > effectiveLocalRetentionMs) {
641642
String message = String.format("Value must not exceed %s (effective value: %d)",
@@ -648,7 +649,7 @@ private static void validateRemoteCopyLagSize(Map<?, ?> props) {
648649
Long retentionBytes = (Long) props.get(TopicConfig.RETENTION_BYTES_CONFIG);
649650
Long localRetentionBytes = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG);
650651
Long remoteCopyLagBytes = (Long) props.get(TopicConfig.REMOTE_COPY_LAG_BYTES_CONFIG);
651-
long effectiveLocalRetentionBytes = localRetentionBytes == -2 ? retentionBytes : localRetentionBytes;
652+
long effectiveLocalRetentionBytes = localRetentionBytes == DEFAULT_LOCAL_RETENTION_BYTES ? retentionBytes : localRetentionBytes;
652653
if (remoteCopyLagBytes > 0 && effectiveLocalRetentionBytes >= 0
653654
&& remoteCopyLagBytes > effectiveLocalRetentionBytes) {
654655
String message = String.format("Value must not exceed %s (effective value: %d)",

0 commit comments

Comments
 (0)