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 2026/10/21 23:03:57 UTC
(kafka) branch trunk updated: KAFKA-15853: [4/N] Move DynamicLogConfig and log config accessors to Java (#23662)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new dbe15eb1b7c KAFKA-15853: [4/N] Move DynamicLogConfig and log config accessors to Java (#23662)
dbe15eb1b7c is described below
commit dbe15eb1b7c987f02993624067f428275238305e
Author: Christo Lolov <lo...@amazon.com>
AuthorDate: Thu Oct 22 00:03:47 2026 +0100
KAFKA-15853: [4/N] Move DynamicLogConfig and log config accessors to Java (#23662)
This PR moves DynamicLogConfig from Scala to Java, along with its
validation of local retention, remote copy lag and cordoned log dirs.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../scala/kafka/server/DynamicBrokerConfig.scala | 115 +--------------
core/src/main/scala/kafka/server/KafkaConfig.scala | 46 +-----
.../kafka/server/DynamicBrokerConfigTest.scala | 2 +-
.../kafka/server/config/AbstractKafkaConfig.java | 50 +++++++
.../kafka/server/config/DynamicBrokerConfig.java | 13 --
.../kafka/server/config/DynamicLogConfig.java | 163 +++++++++++++++++++++
6 files changed, 218 insertions(+), 171 deletions(-)
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 43db686f51d..009033a9677 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -25,7 +25,7 @@ import kafka.network.DataPlaneAcceptor
import kafka.raft.KafkaRaftManager
import kafka.server.DynamicBrokerConfig._
import kafka.utils.Logging
-import org.apache.kafka.common.{Endpoint, Reconfigurable, Uuid}
+import org.apache.kafka.common.{Endpoint, Reconfigurable}
import org.apache.kafka.common.config.{ConfigDef, ConfigException, ConfigResource, SslConfigs}
import org.apache.kafka.common.metadata.{ConfigRecord, MetadataRecordType}
import org.apache.kafka.common.metrics.{Metrics, MetricsReporter}
@@ -38,18 +38,16 @@ import org.apache.kafka.common.utils.internals.ConfigUtils
import org.apache.kafka.network.SocketServer
import org.apache.kafka.raft.{KRaftConfigs, KafkaRaftClient}
import org.apache.kafka.server.{DynamicThreadPool, ProcessRole}
-import org.apache.kafka.server.common.{ApiMessageAndVersion, DirectoryEventHandler}
+import org.apache.kafka.server.common.ApiMessageAndVersion
import org.apache.kafka.server.config.DynamicBrokerConfig.DynamicClusterMirrorConfig
-import org.apache.kafka.server.config.{DynamicConfig, DynamicProducerStateManagerConfig, ServerConfigs, ServerLogConfigs, BrokerReconfigurable => JBrokerReconfigurable, DynamicBrokerConfig => JDynamicBrokerConfig}
+import org.apache.kafka.server.config.{BrokerReconfigurable => JBrokerReconfigurable, DynamicConfig, DynamicLogConfig, DynamicProducerStateManagerConfig, ServerConfigs, DynamicBrokerConfig => JDynamicBrokerConfig}
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.metrics.{ClientTelemetryExporterPlugin, MetricConfigs}
import org.apache.kafka.server.quota.QuotaFactory
import org.apache.kafka.server.telemetry.{ClientTelemetry, ClientTelemetryExporterProvider}
import org.apache.kafka.server.util.LockUtils.{inReadLock, inWriteLock}
import org.apache.kafka.snapshot.RecordsSnapshotReader
-import org.apache.kafka.storage.internals.log.{LogConfig, LogManager}
-import java.util.stream.Collectors
import scala.util.Using
import scala.collection._
import scala.jdk.CollectionConverters._
@@ -546,113 +544,6 @@ trait BrokerReconfigurable {
def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit
}
-class DynamicLogConfig(logManager: LogManager, directoryEventHandler: DirectoryEventHandler) extends BrokerReconfigurable with Logging {
-
- override def reconfigurableConfigs: util.Set[String] = {
- JDynamicBrokerConfig.DynamicLogConfig.RECONFIGURABLE_CONFIGS
- }
-
- override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
- // For update of topic config overrides, only config names and types are validated
- // Names and types have already been validated. For consistency with topic config
- // validation, no additional validation is performed.
-
- def validateLogLocalRetentionMs(): Unit = {
- val logRetentionMs = newConfig.logRetentionTimeMillis
- val logLocalRetentionMs: java.lang.Long = newConfig.remoteLogManagerConfig.logLocalRetentionMs
- if (logRetentionMs != -1L && logLocalRetentionMs != -2L) {
- if (logLocalRetentionMs == -1L) {
- throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs,
- s"Value must not be -1 as ${ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG} value is set as $logRetentionMs.")
- }
- if (logLocalRetentionMs > logRetentionMs) {
- throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs,
- s"Value must not be more than ${ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG} property value: $logRetentionMs")
- }
- }
- }
-
- def validateLogLocalRetentionBytes(): Unit = {
- val logRetentionBytes = newConfig.logRetentionBytes
- val logLocalRetentionBytes: java.lang.Long = newConfig.remoteLogManagerConfig.logLocalRetentionBytes
- if (logRetentionBytes > -1 && logLocalRetentionBytes != -2) {
- if (logLocalRetentionBytes == -1) {
- throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes,
- s"Value must not be -1 as ${ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG} value is set as $logRetentionBytes.")
- }
- if (logLocalRetentionBytes > logRetentionBytes) {
- throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes,
- s"Value must not be more than ${ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG} property value: $logRetentionBytes")
- }
- }
- }
-
- def validateLogRemoteCopyLagMs(): Unit = {
- val logRetentionMs: Long = newConfig.logRetentionTimeMillis
- val logLocalRetentionMs = newConfig.remoteLogManagerConfig.logLocalRetentionMs
- val effectiveLocalRetentionMs = if (logLocalRetentionMs == -2L) logRetentionMs else logLocalRetentionMs
- val logRemoteCopyLagMs = newConfig.remoteLogManagerConfig.logRemoteCopyLagMs
- if (logRemoteCopyLagMs > 0L && effectiveLocalRetentionMs >= 0L && logRemoteCopyLagMs > effectiveLocalRetentionMs) {
- throw new ConfigException(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_MS_PROP, logRemoteCopyLagMs,
- s"Value must not exceed ${RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP} (effective value: $effectiveLocalRetentionMs)")
- }
- }
-
- def validateLogRemoteCopyLagBytes(): Unit = {
- val logRetentionBytes: Long = newConfig.logRetentionBytes
- val logLocalRetentionBytes = newConfig.remoteLogManagerConfig.logLocalRetentionBytes
- val effectiveLocalRetentionBytes = if (logLocalRetentionBytes == -2L) logRetentionBytes else logLocalRetentionBytes
- val logRemoteCopyLagBytes = newConfig.remoteLogManagerConfig.logRemoteCopyLagBytes
- if (logRemoteCopyLagBytes > 0L && effectiveLocalRetentionBytes >= 0L && logRemoteCopyLagBytes > effectiveLocalRetentionBytes) {
- throw new ConfigException(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_BYTES_PROP, logRemoteCopyLagBytes,
- s"Value must not exceed ${RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP} (effective value: $effectiveLocalRetentionBytes)")
- }
- }
-
- def validateCordonedLogDirs(): Unit = {
- val logDirs = newConfig.logDirs()
- val cordonedLogDirs = newConfig.cordonedLogDirs()
- cordonedLogDirs.asScala.foreach(dir =>
- if (!logDirs.contains(dir)) {
- throw new ConfigException(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, cordonedLogDirs, s"Invalid entry in ${ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG}: $dir. " +
- s"All cordoned log dirs must be entries of ${ServerLogConfigs.LOG_DIRS_CONFIG} or ${ServerLogConfigs.LOG_DIR_CONFIG}.")
- }
- )
- }
-
- validateLogLocalRetentionMs()
- validateLogLocalRetentionBytes()
- validateLogRemoteCopyLagMs()
- validateLogRemoteCopyLagBytes()
- validateCordonedLogDirs()
- }
-
- private def updateLogsConfig(newBrokerDefaults: Map[String, Object]): Unit = {
- logManager.brokerConfigUpdated()
- logManager.allLogs.forEach { log =>
- val props = mutable.Map.empty[Any, Any]
- props ++= newBrokerDefaults
- props ++= log.config.originals.asScala.filter { case (k, _) =>
- log.config.overriddenConfigs.contains(k)
- }
-
- val logConfig = new LogConfig(props.asJava, log.config.overriddenConfigs)
- log.updateConfig(logConfig)
- }
- }
-
- override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
- val newBrokerDefaults = new util.HashMap[String, Object](newConfig.extractLogConfigMap)
- logManager.reconfigureDefaultLogConfig(new LogConfig(newBrokerDefaults))
- updateLogsConfig(newBrokerDefaults.asScala)
-
- logManager.updateCordonedLogDirs(util.Set.copyOf(newConfig.cordonedLogDirs))
- directoryEventHandler.handleCordoned(newConfig.cordonedLogDirs.stream
- .flatMap[Uuid](dir => logManager.directoryId(dir).stream)
- .collect(Collectors.toSet[Uuid]))
- }
-}
-
class ControllerDynamicThreadPool(controller: ControllerServer) extends BrokerReconfigurable {
override def reconfigurableConfigs: util.Set[String] = {
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 6d9466084dc..111f387b808 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -21,7 +21,7 @@ import kafka.utils.Implicits._
import kafka.utils.Logging
import org.apache.kafka.common.config.ConfigDef.ConfigKey
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
-import org.apache.kafka.common.config.{ConfigDef, ConfigException, TopicConfig}
+import org.apache.kafka.common.config.{ConfigDef, ConfigException}
import org.apache.kafka.common.internals.Plugin
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.ListenerName
@@ -37,7 +37,6 @@ import org.apache.kafka.security.authorizer.AuthorizerUtils
import org.apache.kafka.server.ProcessRole
import org.apache.kafka.server.authorizer.Authorizer
import org.apache.kafka.server.config.{AbstractKafkaConfig, QuotaConfig, ReplicationConfigs, ServerConfigs, ServerLogConfigs, DynamicBrokerConfig => JDynamicBrokerConfig}
-import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import java.util
import java.util.Properties
@@ -153,9 +152,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
def valuesFromThisConfigWithPrefixOverride(prefix: String): util.Map[String, AnyRef] =
super.valuesWithPrefixOverride(prefix)
- private val _remoteLogManagerConfig = new RemoteLogManagerConfig(this)
- def remoteLogManagerConfig = _remoteLogManagerConfig
-
private val _quorumConfig = new QuorumConfig(this)
def quorumConfig: QuorumConfig = _quorumConfig
@@ -171,7 +167,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
val controllerMaxRecordsPerBatch: Int = getInt(KRaftConfigs.CONTROLLER_MAX_RECORDS_PER_BATCH_CONFIG)
- def maxDecompressedMessageBytes = getInt(ServerConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG)
/************* Authorizer Configuration ***********/
def createNewAuthorizer(metrics: Metrics, role: String): Option[Plugin[Authorizer]] = {
@@ -472,43 +467,4 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
s"${BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG} must implement KafkaPrincipalSerde")
}
- /**
- * Copy the subset of properties that are relevant to Logs. The individual properties
- * are listed here since the names are slightly different in each Config class...
- */
- def extractLogConfigMap: java.util.Map[String, Object] = {
- val logProps = new java.util.HashMap[String, Object]()
- logProps.put(TopicConfig.SEGMENT_BYTES_CONFIG, logSegmentBytes)
- logProps.put(TopicConfig.SEGMENT_MS_CONFIG, logRollTimeMillis)
- logProps.put(TopicConfig.SEGMENT_JITTER_MS_CONFIG, logRollTimeJitterMillis)
- logProps.put(TopicConfig.SEGMENT_INDEX_BYTES_CONFIG, logIndexSizeMaxBytes)
- logProps.put(TopicConfig.FLUSH_MESSAGES_INTERVAL_CONFIG, logFlushIntervalMessages)
- logProps.put(TopicConfig.FLUSH_MS_CONFIG, logFlushIntervalMs)
- logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, logRetentionBytes)
- logProps.put(TopicConfig.RETENTION_MS_CONFIG, logRetentionTimeMillis: java.lang.Long)
- logProps.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, messageMaxBytes)
- logProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, maxDecompressedMessageBytes)
- logProps.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, logIndexIntervalBytes)
- logProps.put(TopicConfig.DELETE_RETENTION_MS_CONFIG, logCleanerDeleteRetentionMs)
- logProps.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, logCleanerMinCompactionLagMs)
- logProps.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, logCleanerMaxCompactionLagMs)
- logProps.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, logDeleteDelayMs)
- logProps.put(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, logCleanerMinCleanRatio)
- logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, logCleanupPolicy)
- logProps.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, minInSyncReplicas)
- logProps.put(TopicConfig.COMPRESSION_TYPE_CONFIG, compressionType)
- logProps.put(TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG, gzipCompressionLevel)
- logProps.put(TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG, lz4CompressionLevel)
- logProps.put(TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG, zstdCompressionLevel)
- logProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, uncleanLeaderElectionEnable)
- logProps.put(TopicConfig.PREALLOCATE_CONFIG, logPreAllocateEnable)
- logProps.put(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, logMessageTimestampType.name)
- logProps.put(TopicConfig.MESSAGE_TIMESTAMP_BEFORE_MAX_MS_CONFIG, logMessageTimestampBeforeMaxMs: java.lang.Long)
- logProps.put(TopicConfig.MESSAGE_TIMESTAMP_AFTER_MAX_MS_CONFIG, logMessageTimestampAfterMaxMs: java.lang.Long)
- logProps.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, remoteLogManagerConfig.logLocalRetentionMs: java.lang.Long)
- logProps.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, remoteLogManagerConfig.logLocalRetentionBytes: java.lang.Long)
- logProps.put(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG, remoteLogManagerConfig.logRemoteCopyLagMs: java.lang.Long)
- logProps.put(TopicConfig.REMOTE_COPY_LAG_BYTES_CONFIG, remoteLogManagerConfig.logRemoteCopyLagBytes: java.lang.Long)
- logProps
- }
}
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index ee721a4151c..4821176e593 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -33,7 +33,7 @@ import org.apache.kafka.raft.{KRaftConfigs, QuorumConfig}
import org.apache.kafka.server.DynamicThreadPool
import org.apache.kafka.server.authorizer._
import org.apache.kafka.server.common.DirectoryEventHandler
-import org.apache.kafka.server.config.{ReplicationConfigs, ServerConfigs, ServerLogConfigs}
+import org.apache.kafka.server.config.{DynamicLogConfig, ReplicationConfigs, ServerConfigs, ServerLogConfigs}
import org.apache.kafka.server.log.remote.storage.{RemoteLogManager, RemoteLogManagerConfig}
import org.apache.kafka.server.metrics.{ClientTelemetryExporterPlugin, KafkaYammerMetrics, MetricConfigs}
import org.apache.kafka.server.quota.QuotaFactory
diff --git a/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java b/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java
index e848908e894..d61a4d84d2e 100644
--- a/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java
+++ b/server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java
@@ -23,6 +23,7 @@ import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.network.ListenerName;
@@ -104,6 +105,51 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
super(definition, originals, configProviderProps, doLog);
}
+ public RemoteLogManagerConfig remoteLogManagerConfig() {
+ return new RemoteLogManagerConfig(this);
+ }
+
+ /**
+ * Copy the subset of properties that are relevant to Logs. The individual properties
+ * are listed here since the names are slightly different in each Config class...
+ */
+ public Map<String, Object> extractLogConfigMap() {
+ Map<String, Object> logProps = new HashMap<>();
+ logProps.put(TopicConfig.SEGMENT_BYTES_CONFIG, logSegmentBytes());
+ logProps.put(TopicConfig.SEGMENT_MS_CONFIG, logRollTimeMillis());
+ logProps.put(TopicConfig.SEGMENT_JITTER_MS_CONFIG, logRollTimeJitterMillis());
+ logProps.put(TopicConfig.SEGMENT_INDEX_BYTES_CONFIG, logIndexSizeMaxBytes());
+ logProps.put(TopicConfig.FLUSH_MESSAGES_INTERVAL_CONFIG, logFlushIntervalMessages());
+ logProps.put(TopicConfig.FLUSH_MS_CONFIG, logFlushIntervalMs());
+ logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, logRetentionBytes());
+ logProps.put(TopicConfig.RETENTION_MS_CONFIG, logRetentionTimeMillis());
+ logProps.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, messageMaxBytes());
+ logProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, maxDecompressedMessageBytes());
+ logProps.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, logIndexIntervalBytes());
+ logProps.put(TopicConfig.DELETE_RETENTION_MS_CONFIG, logCleanerDeleteRetentionMs());
+ logProps.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, logCleanerMinCompactionLagMs());
+ logProps.put(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, logCleanerMaxCompactionLagMs());
+ logProps.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, logDeleteDelayMs());
+ logProps.put(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, logCleanerMinCleanRatio());
+ logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, logCleanupPolicy());
+ logProps.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, minInSyncReplicas());
+ logProps.put(TopicConfig.COMPRESSION_TYPE_CONFIG, compressionType());
+ logProps.put(TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG, gzipCompressionLevel());
+ logProps.put(TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG, lz4CompressionLevel());
+ logProps.put(TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG, zstdCompressionLevel());
+ logProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, uncleanLeaderElectionEnable());
+ logProps.put(TopicConfig.PREALLOCATE_CONFIG, logPreAllocateEnable());
+ logProps.put(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, logMessageTimestampType().name);
+ logProps.put(TopicConfig.MESSAGE_TIMESTAMP_BEFORE_MAX_MS_CONFIG, logMessageTimestampBeforeMaxMs());
+ logProps.put(TopicConfig.MESSAGE_TIMESTAMP_AFTER_MAX_MS_CONFIG, logMessageTimestampAfterMaxMs());
+ RemoteLogManagerConfig remoteLogManagerConfig = remoteLogManagerConfig();
+ logProps.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, remoteLogManagerConfig.logLocalRetentionMs());
+ logProps.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, remoteLogManagerConfig.logLocalRetentionBytes());
+ logProps.put(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG, remoteLogManagerConfig.logRemoteCopyLagMs());
+ logProps.put(TopicConfig.REMOTE_COPY_LAG_BYTES_CONFIG, remoteLogManagerConfig.logRemoteCopyLagBytes());
+ return Collections.unmodifiableMap(logProps);
+ }
+
public List<String> logDirs() {
return Optional.ofNullable(getList(ServerLogConfigs.LOG_DIRS_CONFIG)).orElse(getList(ServerLogConfigs.LOG_DIR_CONFIG));
}
@@ -738,6 +784,10 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
}
+ public Integer maxDecompressedMessageBytes() {
+ return getInt(ServerConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG);
+ }
+
public int getNumReplicaAlterLogDirsThreads() {
Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
return numThreads != null ? numThreads : logDirs().size();
diff --git a/server/src/main/java/org/apache/kafka/server/config/DynamicBrokerConfig.java b/server/src/main/java/org/apache/kafka/server/config/DynamicBrokerConfig.java
index e5df7b6298a..f639deab1fe 100644
--- a/server/src/main/java/org/apache/kafka/server/config/DynamicBrokerConfig.java
+++ b/server/src/main/java/org/apache/kafka/server/config/DynamicBrokerConfig.java
@@ -214,19 +214,6 @@ public class DynamicBrokerConfig {
);
}
- public static class DynamicLogConfig {
- /**
- * The broker configurations pertaining to logs that are reconfigurable. This set contains
- * the names you would use when setting a static or dynamic broker configuration (not topic
- * configuration).
- */
- public static final Set<String> RECONFIGURABLE_CONFIGS = Stream.of(
- ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.values(),
- Set.of(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG))
- .flatMap(Collection::stream)
- .collect(Collectors.toUnmodifiableSet());
- }
-
public static class DynamicListenerConfig {
/**
* The set of configurations which the DynamicListenerConfig object listens for. Many of
diff --git a/server/src/main/java/org/apache/kafka/server/config/DynamicLogConfig.java b/server/src/main/java/org/apache/kafka/server/config/DynamicLogConfig.java
new file mode 100644
index 00000000000..0566499e561
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/config/DynamicLogConfig.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.config;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.server.common.DirectoryEventHandler;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
+import org.apache.kafka.storage.internals.log.LogConfig;
+import org.apache.kafka.storage.internals.log.LogManager;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class DynamicLogConfig implements BrokerReconfigurable {
+
+ /**
+ * The broker configurations pertaining to logs that are reconfigurable. This set contains
+ * the names you would use when setting a static or dynamic broker configuration (not topic
+ * configuration).
+ */
+ public static final Set<String> RECONFIGURABLE_CONFIGS = Stream.of(
+ ServerTopicConfigSynonyms.TOPIC_CONFIG_SYNONYMS.values(),
+ Set.of(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG))
+ .flatMap(Collection::stream)
+ .collect(Collectors.toUnmodifiableSet());
+
+ private final LogManager logManager;
+ private final DirectoryEventHandler directoryEventHandler;
+
+ public DynamicLogConfig(LogManager logManager, DirectoryEventHandler directoryEventHandler) {
+ this.logManager = logManager;
+ this.directoryEventHandler = directoryEventHandler;
+ }
+
+ @Override
+ public Set<String> reconfigurableConfigs() {
+ return RECONFIGURABLE_CONFIGS;
+ }
+
+ @Override
+ public void validateReconfiguration(AbstractKafkaConfig newConfig) {
+ // For update of topic config overrides, only config names and types are validated
+ // Names and types have already been validated. For consistency with topic config
+ // validation, no additional validation is performed.
+ validateLogLocalRetentionMs(newConfig);
+ validateLogLocalRetentionBytes(newConfig);
+ validateLogRemoteCopyLagMs(newConfig);
+ validateLogRemoteCopyLagBytes(newConfig);
+ validateCordonedLogDirs(newConfig);
+ }
+
+ private void validateLogLocalRetentionMs(AbstractKafkaConfig newConfig) {
+ long logRetentionMs = newConfig.logRetentionTimeMillis();
+ long logLocalRetentionMs = newConfig.remoteLogManagerConfig().logLocalRetentionMs();
+ if (logRetentionMs != -1L && logLocalRetentionMs != -2L) {
+ if (logLocalRetentionMs == -1L) {
+ throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs,
+ "Value must not be -1 as " + ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG + " value is set as " + logRetentionMs + ".");
+ }
+ if (logLocalRetentionMs > logRetentionMs) {
+ throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, logLocalRetentionMs,
+ "Value must not be more than " + ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG + " property value: " + logRetentionMs);
+ }
+ }
+ }
+
+ private void validateLogLocalRetentionBytes(AbstractKafkaConfig newConfig) {
+ long logRetentionBytes = newConfig.logRetentionBytes();
+ long logLocalRetentionBytes = newConfig.remoteLogManagerConfig().logLocalRetentionBytes();
+ if (logRetentionBytes > -1L && logLocalRetentionBytes != -2L) {
+ if (logLocalRetentionBytes == -1L) {
+ throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes,
+ "Value must not be -1 as " + ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG + " value is set as " + logRetentionBytes + ".");
+ }
+ if (logLocalRetentionBytes > logRetentionBytes) {
+ throw new ConfigException(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, logLocalRetentionBytes,
+ "Value must not be more than " + ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG + " property value: " + logRetentionBytes);
+ }
+ }
+ }
+
+ private void validateLogRemoteCopyLagMs(AbstractKafkaConfig newConfig) {
+ RemoteLogManagerConfig remoteLogManagerConfig = newConfig.remoteLogManagerConfig();
+ long logRetentionMs = newConfig.logRetentionTimeMillis();
+ long logLocalRetentionMs = remoteLogManagerConfig.logLocalRetentionMs();
+ long effectiveLocalRetentionMs = logLocalRetentionMs == -2L ? logRetentionMs : logLocalRetentionMs;
+ long logRemoteCopyLagMs = remoteLogManagerConfig.logRemoteCopyLagMs();
+ if (logRemoteCopyLagMs > 0L && effectiveLocalRetentionMs >= 0L && logRemoteCopyLagMs > effectiveLocalRetentionMs) {
+ throw new ConfigException(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_MS_PROP, logRemoteCopyLagMs,
+ "Value must not exceed " + RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP + " (effective value: " + effectiveLocalRetentionMs + ")");
+ }
+ }
+
+ private void validateLogRemoteCopyLagBytes(AbstractKafkaConfig newConfig) {
+ RemoteLogManagerConfig remoteLogManagerConfig = newConfig.remoteLogManagerConfig();
+ long logRetentionBytes = newConfig.logRetentionBytes();
+ long logLocalRetentionBytes = remoteLogManagerConfig.logLocalRetentionBytes();
+ long effectiveLocalRetentionBytes = logLocalRetentionBytes == -2L ? logRetentionBytes : logLocalRetentionBytes;
+ long logRemoteCopyLagBytes = remoteLogManagerConfig.logRemoteCopyLagBytes();
+ if (logRemoteCopyLagBytes > 0L && effectiveLocalRetentionBytes >= 0L && logRemoteCopyLagBytes > effectiveLocalRetentionBytes) {
+ throw new ConfigException(RemoteLogManagerConfig.LOG_REMOTE_COPY_LAG_BYTES_PROP, logRemoteCopyLagBytes,
+ "Value must not exceed " + RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP + " (effective value: " + effectiveLocalRetentionBytes + ")");
+ }
+ }
+
+ private void validateCordonedLogDirs(AbstractKafkaConfig newConfig) {
+ List<String> logDirs = newConfig.logDirs();
+ List<String> cordonedLogDirs = newConfig.cordonedLogDirs();
+ cordonedLogDirs.forEach(dir -> {
+ if (!logDirs.contains(dir)) {
+ throw new ConfigException(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, cordonedLogDirs,
+ "Invalid entry in " + ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG + ": " + dir + ". " +
+ "All cordoned log dirs must be entries of " + ServerLogConfigs.LOG_DIRS_CONFIG + " or " + ServerLogConfigs.LOG_DIR_CONFIG + ".");
+ }
+ });
+ }
+
+ private void updateLogsConfig(Map<String, Object> newBrokerDefaults) {
+ logManager.brokerConfigUpdated();
+ logManager.allLogs().forEach(log -> {
+ Map<String, Object> props = new HashMap<>(newBrokerDefaults);
+ LogConfig logConfig = log.config();
+ logConfig.originals().forEach((k, v) -> {
+ if (logConfig.overriddenConfigs.contains(k)) {
+ props.put(k, v);
+ }
+ });
+ log.updateConfig(new LogConfig(props, logConfig.overriddenConfigs));
+ });
+ }
+
+ @Override
+ public void reconfigure(AbstractKafkaConfig oldConfig, AbstractKafkaConfig newConfig) {
+ Map<String, Object> newBrokerDefaults = newConfig.extractLogConfigMap();
+ logManager.reconfigureDefaultLogConfig(new LogConfig(newBrokerDefaults));
+ updateLogsConfig(newBrokerDefaults);
+
+ List<String> cordonedLogDirs = newConfig.cordonedLogDirs();
+ logManager.updateCordonedLogDirs(Set.copyOf(cordonedLogDirs));
+ directoryEventHandler.handleCordoned(cordonedLogDirs.stream()
+ .flatMap(dir -> logManager.directoryId(dir).stream())
+ .collect(Collectors.toUnmodifiableSet()));
+ }
+}