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/09/23 11:05:00 UTC
(kafka) branch trunk updated: KAFKA-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig (#21943)
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 e2177c008e4 KAFKA-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig (#21943)
e2177c008e4 is described below
commit e2177c008e45ca459608a3476c64bd8f4084dc19
Author: Christo Lolov <lo...@amazon.com>
AuthorDate: Wed Sep 23 12:04:53 2026 +0100
KAFKA-15853: [2/N] Move bulk of remaining configs to AbstractKafkaConfig (#21943)
This PR moves the bulk of the remaining KafkaConfig accessors from Scala
to Java. The Scala vals become Java methods that read the value on each
call; since none of them are dynamically reconfigurable, behavior is
unchanged. The validation of group.coordinator.rebalance.protocols is
moved into validateValues, and the test-only isKRaftCombinedMode is
removed.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Ming-Yen Chung
<mi...@gmail.com>
---
.../main/scala/kafka/raft/KafkaRaftManager.scala | 2 +-
.../main/scala/kafka/server/ControllerServer.scala | 4 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 169 +-----------
.../main/scala/kafka/server/ReplicaManager.scala | 4 +-
core/src/main/scala/kafka/tools/StorageTool.scala | 2 +-
.../test/scala/unit/kafka/KafkaConfigTest.scala | 3 +-
.../scala/unit/kafka/server/KafkaConfigTest.scala | 6 +-
.../kafka/server/config/AbstractKafkaConfig.java | 290 +++++++++++++++++++++
8 files changed, 314 insertions(+), 166 deletions(-)
diff --git a/core/src/main/scala/kafka/raft/KafkaRaftManager.scala b/core/src/main/scala/kafka/raft/KafkaRaftManager.scala
index 65c2d0651a9..9a942304ec8 100644
--- a/core/src/main/scala/kafka/raft/KafkaRaftManager.scala
+++ b/core/src/main/scala/kafka/raft/KafkaRaftManager.scala
@@ -117,7 +117,7 @@ class KafkaRaftManager[T](
val differentMetadataLogDir = KafkaRaftManager.hasDifferentLogDir(config)
// Or this node is only a controller
- val isOnlyController = config.processRoles == Set(ProcessRole.ControllerRole)
+ val isOnlyController = config.processRoles.asScala == Set(ProcessRole.ControllerRole)
if (differentMetadataLogDir || isOnlyController) {
Some(KafkaRaftManager.lockDataDir(new File(config.metadataLogDir)))
diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala
index 625a0b9a4c4..64a6ffdfb5e 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/core/src/main/scala/kafka/server/ControllerServer.scala
@@ -235,8 +235,6 @@ class ControllerServer(
OptionalLong.empty()
}
- val maxIdleIntervalNs = config.metadataMaxIdleIntervalNs.fold(OptionalLong.empty)(OptionalLong.of)
-
quorumControllerMetrics = new QuorumControllerMetrics(Optional.of(KafkaYammerMetrics.defaultRegistry), time, config.brokerSessionTimeoutMs)
new QuorumController.Builder(config.nodeId, sharedServer.clusterId).
@@ -250,7 +248,7 @@ class ControllerServer(
setSessionTimeoutNs(TimeUnit.NANOSECONDS.convert(config.brokerSessionTimeoutMs.longValue(),
TimeUnit.MILLISECONDS)).
setLeaderImbalanceCheckIntervalNs(leaderImbalanceCheckIntervalNs).
- setMaxIdleIntervalNs(maxIdleIntervalNs).
+ setMaxIdleIntervalNs(config.metadataMaxIdleIntervalNs).
setMetrics(quorumControllerMetrics).
setCreateTopicPolicy(createTopicPolicy.toJava).
setAlterConfigPolicy(alterConfigPolicy.toJava).
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index a9bfc9e6b06..00cba8e4e7d 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -17,36 +17,32 @@
package kafka.server
-import java.util
-import java.util.concurrent.TimeUnit
-import java.util.Properties
-import kafka.utils.Logging
import kafka.utils.Implicits._
-import org.apache.kafka.common.{Endpoint, Reconfigurable}
-import org.apache.kafka.common.config.{ConfigDef, ConfigException, TopicConfig}
+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.internals.Plugin
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.ListenerName
-import org.apache.kafka.common.security.auth.KafkaPrincipalSerde
-import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.security.auth.{KafkaPrincipalSerde, SecurityProtocol}
import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.coordinator.group.Group.GroupType
-import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig
+import org.apache.kafka.common.{Endpoint, Reconfigurable}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
+import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig
import org.apache.kafka.coordinator.share.ShareCoordinatorConfig
import org.apache.kafka.network.SocketServerConfigs
-import org.apache.kafka.raft.{KRaftConfigs, MetadataLogConfig, QuorumConfig}
+import org.apache.kafka.raft.{KRaftConfigs, QuorumConfig}
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 org.apache.kafka.server.metrics.MetricConfigs
-import scala.jdk.CollectionConverters._
+import java.util
+import java.util.Properties
import scala.collection.{Map, Seq}
+import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters.{RichOption, RichOptional}
object KafkaConfig extends Logging {
@@ -125,7 +121,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
// Cache the current config to avoid acquiring read lock to access from dynamicConfig
@volatile private var currentConfig = this
- val processRoles: Set[ProcessRole] = parseProcessRoles()
private[server] val dynamicConfig = new DynamicBrokerConfig(this)
private[server] def updateCurrentConfig(newConfig: KafkaConfig): Unit = {
@@ -174,52 +169,10 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
private val _shareCoordinatorConfig = new ShareCoordinatorConfig(this)
def shareCoordinatorConfig: ShareCoordinatorConfig = _shareCoordinatorConfig
- /** ********* General Configuration ***********/
- val brokerSessionTimeoutMs: Int = getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG)
- val controllerPerformanceSamplePeriodMs: Long = getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS)
- val controllerPerformanceAlwaysLogThresholdMs: Long = getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS)
val controllerMaxRecordsPerBatch: Int = getInt(KRaftConfigs.CONTROLLER_MAX_RECORDS_PER_BATCH_CONFIG)
- private def parseProcessRoles(): Set[ProcessRole] = {
- val roles = getList(KRaftConfigs.PROCESS_ROLES_CONFIG).asScala.map {
- case "broker" => ProcessRole.BrokerRole
- case "controller" => ProcessRole.ControllerRole
- case role => throw new ConfigException(s"Unknown process role '$role'" +
- " (only 'broker' and 'controller' are allowed roles)")
- }
- roles.toSet
- }
-
- def isKRaftCombinedMode: Boolean = {
- processRoles == Set(ProcessRole.BrokerRole, ProcessRole.ControllerRole)
- }
-
- def metadataLogDir: String = {
- Option(getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG)) match {
- case Some(dir) => dir
- case None => logDirs.get(0)
- }
- }
-
- val serverMaxStartupTimeMs = getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG)
-
- def messageMaxBytes = getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG)
-
def maxDecompressedMessageBytes = getInt(ServerConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG)
- def getNumReplicaAlterLogDirsThreads: Int = {
- val numThreads: Integer = Option(getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG)).getOrElse(logDirs.size)
- numThreads
- }
-
- /************* Metadata Configuration ***********/
- val metadataSnapshotMaxNewRecordBytes = getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG)
- val metadataSnapshotMaxIntervalMs = getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG)
- val metadataMaxIdleIntervalNs: Option[Long] = {
- val value = TimeUnit.NANOSECONDS.convert(getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).toLong, TimeUnit.MILLISECONDS)
- if (value > 0) Some(value) else None
- }
-
/************* Authorizer Configuration ***********/
def createNewAuthorizer(metrics: Metrics, role: String): Option[Plugin[Authorizer]] = {
val className = getString(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG)
@@ -247,101 +200,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
}
}
- /***************** rack configuration **************/
- val replicaSelectorClassName = Option(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG))
-
- /** ********* Replication configuration ***********/
- val replicaLagTimeMaxMs = getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG)
- val replicaSocketTimeoutMs = getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG)
- val replicaSocketReceiveBufferBytes = getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG)
- val replicaFetchMaxBytes = getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG)
- val replicaFetchWaitMaxMs = getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG)
- val replicaFetchMinBytes = getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG)
- val replicaFetchResponseMaxBytes = getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG)
- val replicaFetchBackoffMs = getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG)
- val replicaHighWatermarkCheckpointIntervalMs = getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG)
- val fetchPurgatoryPurgeIntervalRequests = getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG)
- val producerPurgatoryPurgeIntervalRequests = getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG)
- val deleteRecordsPurgatoryPurgeIntervalRequests = getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG)
- val autoLeaderRebalanceEnable = getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG)
- val leaderImbalanceCheckIntervalSeconds: Long = getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG)
- val uncleanLeaderElectionCheckIntervalMs: Long = getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG)
- def uncleanLeaderElectionEnable: java.lang.Boolean = getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG)
- def followerFetchLastTieredOffsetEnable: java.lang.Boolean = getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
-
- /** ********* Controlled shutdown configuration ***********/
- val controlledShutdownEnable = getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG)
-
- /** Group coordinator configs */
- val groupCoordinatorRebalanceProtocols = {
- val protocols = getList(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)
- .asScala.map(_.toUpperCase).map(GroupType.valueOf).toSet
- if (!protocols.contains(GroupType.CLASSIC)) {
- throw new ConfigException(s"Disabling the '${GroupType.CLASSIC}' protocol is not supported.")
- }
- if (doLog && protocols.contains(GroupType.SHARE)) {
- warn(s"'${GroupType.SHARE}' in `${GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG}` is deprecated. " +
- s"Share groups are controlled by the 'share.version' feature. " +
- s"This config will be removed in Kafka 5.0.")
- }
- if (doLog && originals().containsKey(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)) {
- val defaultProtocols = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT
- .asScala.map(_.toUpperCase).map(GroupType.valueOf).toSet
- val missingProtocols = defaultProtocols -- protocols
- if (missingProtocols.nonEmpty) {
- warn(s"The config `${GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG}` is deprecated and will be removed in Kafka 5.0. " +
- s"The following protocol(s) are currently disabled: ${missingProtocols.mkString(", ")}. " +
- s"In Kafka 5.0, all protocols will always be enabled and controlled solely by feature versions " +
- s"(group.version, streams.version, share.version) via kafka-features.sh. " +
- s"Please remove the configuration, which will restore all protocols to the default enabled state, to prepare for the upgrade.")
- } else {
- warn(s"The config `${GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG}` is deprecated and will be removed in Kafka 5.0. " +
- s"Please remove the configuration to prepare for the upgrade.")
- }
- }
- protocols
- }
-
- /** ********* Metric Configuration **************/
- val metricNumSamples = getInt(MetricConfigs.METRIC_NUM_SAMPLES_CONFIG)
- val metricSampleWindowMs = getLong(MetricConfigs.METRIC_SAMPLE_WINDOW_MS_CONFIG)
- val metricRecordingLevel = getString(MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG)
-
- /** ********* Kafka Client Telemetry Metrics Configuration ***********/
- val clientTelemetryMaxBytes: Int = getInt(MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_CONFIG)
-
- /** ********* SSL/SASL Configuration **************/
- // Security configs may be overridden for listeners, so it is not safe to use the base values
- // Hence the base SSL/SASL configs are not fields of KafkaConfig, listener configs should be
- // retrieved using KafkaConfig#valuesWithPrefixOverride
- private def saslEnabledMechanisms(listenerName: ListenerName): Set[String] = {
- val value = valuesWithPrefixOverride(listenerName.configPrefix).get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG)
- if (value != null)
- value.asInstanceOf[util.List[String]].asScala.toSet
- else
- Set.empty[String]
- }
-
- def saslMechanismInterBrokerProtocol = getString(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG)
-
- /** ********* Fetch Configuration **************/
- val maxIncrementalFetchSessionCacheSlots = getInt(ServerConfigs.MAX_INCREMENTAL_FETCH_SESSION_CACHE_SLOTS_CONFIG)
- val fetchMaxBytes = getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG)
-
- /** ********* Request Limit Configuration ***********/
- val maxRequestPartitionSizeLimit = getInt(ServerConfigs.MAX_REQUEST_PARTITION_SIZE_LIMIT_CONFIG)
-
- val deleteTopicEnable = getBoolean(ServerConfigs.DELETE_TOPIC_ENABLE_CONFIG)
- def compressionType = getString(ServerConfigs.COMPRESSION_TYPE_CONFIG)
-
- def gzipCompressionLevel = getInt(ServerConfigs.COMPRESSION_GZIP_LEVEL_CONFIG)
- def lz4CompressionLevel = getInt(ServerConfigs.COMPRESSION_LZ4_LEVEL_CONFIG)
- def zstdCompressionLevel = getInt(ServerConfigs.COMPRESSION_ZSTD_LEVEL_CONFIG)
-
- /** Internal Configurations **/
- val unstableApiVersionsEnabled = getBoolean(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG)
- val unstableFeatureVersionsEnabled = getBoolean(ServerConfigs.UNSTABLE_FEATURE_VERSIONS_ENABLE_CONFIG)
-
override def addReconfigurable(reconfigurable: Reconfigurable): Unit = {
dynamicConfig.addReconfigurable(reconfigurable)
}
@@ -444,6 +302,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
validateValues()
private def validateValues(): Unit = {
+ validateGroupCoordinatorRebalanceProtocols(doLog)
if (nodeId != brokerId) {
throw new ConfigException(s"`${KRaftConfigs.NODE_ID_CONFIG}` and `${ServerConfigs.BROKER_ID_CONFIG}` must be set to the same value. " +
s"`${ServerConfigs.BROKER_ID_CONFIG}` is deprecated, please use `${KRaftConfigs.NODE_ID_CONFIG}` instead.")
@@ -499,10 +358,10 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
}
def warnIfConfigDefinedInWrongRole(expectedRole: ProcessRole, configName: String, extraMessage: String = ""): Unit = {
if (originals.containsKey(configName)) {
- warn(s"$configName is defined in ${processRoles.mkString(", ")}. It should be defined in the $expectedRole role. $extraMessage")
+ warn(s"$configName is defined in ${processRoles.asScala.mkString(", ")}. It should be defined in the $expectedRole role. $extraMessage")
}
}
- if (processRoles == Set(ProcessRole.BrokerRole)) {
+ if (processRoles.asScala == Set(ProcessRole.BrokerRole)) {
// KRaft broker-only
validateQuorumVotersAndQuorumBootstrapServerForKRaft()
// nodeId must not appear in controller.quorum.voters
@@ -536,7 +395,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
val ignoredIn5 = "This configuration will be ignored in the broker role in 5.0."
warnIfConfigDefinedInWrongRole(ProcessRole.ControllerRole, ServerLogConfigs.NUM_PARTITIONS_CONFIG, ignoredIn5)
warnIfConfigDefinedInWrongRole(ProcessRole.ControllerRole, ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, ignoredIn5)
- } else if (processRoles == Set(ProcessRole.ControllerRole)) {
+ } else if (processRoles.asScala == Set(ProcessRole.ControllerRole)) {
// KRaft controller-only
validateQuorumVotersAndQuorumBootstrapServerForKRaft()
// listeners should only contain listeners also enumerated in the controller listener
@@ -554,7 +413,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
validateControllerQuorumVotersMustContainNodeIdForKRaftController()
validateAdvertisedControllerListenersNonEmptyForKRaftController()
validateControllerListenerNamesMustAppearInListenersForKRaftController()
- } else if (isKRaftCombinedMode) {
+ } else if (processRoles == util.Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole)) {
// KRaft combined broker and controller
validateQuorumVotersAndQuorumBootstrapServerForKRaft()
validateControllerQuorumVotersMustContainNodeIdForKRaftController()
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index ec219718cbc..e7e907b4fee 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -2194,7 +2194,7 @@ class ReplicaManager(val config: KafkaConfig,
s"for partitions ${partitionsWithOfflineFutureReplica.mkString(",")} because they are in the failed log directory $dir.")
}
logManager.handleLogDirFailure(dir)
- if (dir == new File(config.metadataLogDir).getAbsolutePath && config.processRoles.nonEmpty) {
+ if (dir == new File(config.metadataLogDir).getAbsolutePath && !config.processRoles.isEmpty) {
fatal(s"Shutdown broker because the metadata log dir $dir has failed")
Exit.halt(1)
}
@@ -2257,7 +2257,7 @@ class ReplicaManager(val config: KafkaConfig,
}
private def createReplicaSelector(metrics: Metrics): Option[Plugin[ReplicaSelector]] = {
- config.replicaSelectorClassName.map { className =>
+ config.replicaSelectorClassName.toScala.map { className =>
val tmpReplicaSelector: ReplicaSelector = Utils.newInstance(className, classOf[ReplicaSelector])
tmpReplicaSelector.configure(config.originals())
Plugin.wrapInstance(tmpReplicaSelector, metrics, ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG)
diff --git a/core/src/main/scala/kafka/tools/StorageTool.scala b/core/src/main/scala/kafka/tools/StorageTool.scala
index 23df78805e4..3eadfc378ee 100644
--- a/core/src/main/scala/kafka/tools/StorageTool.scala
+++ b/core/src/main/scala/kafka/tools/StorageTool.scala
@@ -82,7 +82,7 @@ object StorageTool extends Logging {
command match {
case "info" =>
val directories = configToLogDirectories(config.get)
- infoCommand(printStream, config.get.processRoles.nonEmpty, directories)
+ infoCommand(printStream, !config.get.processRoles.isEmpty, directories)
case "format" =>
runFormatCommand(namespace, config.get, printStream)
diff --git a/core/src/test/scala/unit/kafka/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
index ad6387c6e39..96221fc7adf 100644
--- a/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/KafkaConfigTest.scala
@@ -29,6 +29,7 @@ import org.apache.kafka.common.internals.FatalExitError
import org.apache.kafka.common.utils.internals.Exit
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.raft.{KRaftConfigs, QuorumConfig}
+import org.apache.kafka.server.ProcessRole
import org.apache.kafka.server.config.ReplicationConfigs
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.api.Assertions._
@@ -221,7 +222,7 @@ class KafkaConfigTest {
propertiesFile.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "1@localhost:9092")
setListenerProps(propertiesFile)
val config = KafkaConfig.fromProps(propertiesFile)
- assertTrue(config.isKRaftCombinedMode)
+ assertTrue(config.processRoles() == util.Set.of(ProcessRole.BrokerRole, ProcessRole.ControllerRole))
}
@Test
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 1718e728d07..9d6ac89a7ca 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1939,16 +1939,16 @@ class KafkaConfigTest {
// This is OK.
props.put(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer")
val config = KafkaConfig.fromProps(props)
- assertEquals(Set(GroupType.CLASSIC, GroupType.CONSUMER), config.groupCoordinatorRebalanceProtocols)
+ assertEquals(Set(GroupType.CLASSIC, GroupType.CONSUMER), config.groupCoordinatorRebalanceProtocols.asScala.toSet)
props.put(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,streams")
val config2 = KafkaConfig.fromProps(props)
- assertEquals(Set(GroupType.CLASSIC, GroupType.STREAMS), config2.groupCoordinatorRebalanceProtocols)
+ assertEquals(Set(GroupType.CLASSIC, GroupType.STREAMS), config2.groupCoordinatorRebalanceProtocols.asScala.toSet)
// Including "share" is also OK
props.put(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,share")
val config3 = KafkaConfig.fromProps(props)
- assertEquals(Set(GroupType.CLASSIC, GroupType.CONSUMER, GroupType.SHARE), config3.groupCoordinatorRebalanceProtocols)
+ assertEquals(Set(GroupType.CLASSIC, GroupType.CONSUMER, GroupType.SHARE), config3.groupCoordinatorRebalanceProtocols.asScala.toSet)
}
@Test
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 1d24e946713..76b24707f6f 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
@@ -29,6 +29,7 @@ import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.Group.GroupType;
import org.apache.kafka.coordinator.group.GroupConfig;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig;
@@ -40,6 +41,7 @@ import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.raft.KRaftConfigs;
import org.apache.kafka.raft.MetadataLogConfig;
import org.apache.kafka.raft.QuorumConfig;
+import org.apache.kafka.server.ProcessRole;
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
import org.apache.kafka.server.metrics.MetricConfigs;
import org.apache.kafka.server.util.Csv;
@@ -47,12 +49,17 @@ import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.commons.validator.routines.InetAddressValidator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -65,6 +72,8 @@ import java.util.stream.Collectors;
*/
public abstract class AbstractKafkaConfig extends AbstractConfig {
+ private static final Logger log = LoggerFactory.getLogger(AbstractKafkaConfig.class);
+
private static final InetAddressValidator INET_ADDRESS_VALIDATOR = InetAddressValidator.getInstance();
public static final ConfigDef CONFIG_DEF = Utils.mergeConfigs(List.of(
@@ -670,4 +679,285 @@ public abstract class AbstractKafkaConfig extends AbstractConfig {
);
return defaults;
}
+
+ // ********* General Configuration **********
+
+ public int brokerSessionTimeoutMs() {
+ return getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG);
+ }
+
+ public long controllerPerformanceSamplePeriodMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS);
+ }
+
+ public long controllerPerformanceAlwaysLogThresholdMs() {
+ return getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS);
+ }
+
+ public Set<ProcessRole> processRoles() {
+ Set<ProcessRole> result = new HashSet<>();
+ for (String role : getList(KRaftConfigs.PROCESS_ROLES_CONFIG)) {
+ switch (role) {
+ case "broker" -> result.add(ProcessRole.BrokerRole);
+ case "controller" -> result.add(ProcessRole.ControllerRole);
+ default -> throw new ConfigException("Unknown process role '" + role +
+ "' (only 'broker' and 'controller' are allowed roles)");
+ }
+ }
+ return Collections.unmodifiableSet(result);
+ }
+
+ public String metadataLogDir() {
+ String dir = getString(MetadataLogConfig.METADATA_LOG_DIR_CONFIG);
+ return dir != null ? dir : logDirs().get(0);
+ }
+
+ public long serverMaxStartupTimeMs() {
+ return getLong(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
+ }
+
+ public Integer messageMaxBytes() {
+ return getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ }
+
+ public int getNumReplicaAlterLogDirsThreads() {
+ Integer numThreads = getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG);
+ return numThreads != null ? numThreads : logDirs().size();
+ }
+
+ // ********* Metadata Configuration **********
+
+ public long metadataSnapshotMaxNewRecordBytes() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG);
+ }
+
+ public long metadataSnapshotMaxIntervalMs() {
+ return getLong(MetadataLogConfig.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG);
+ }
+
+ public OptionalLong metadataMaxIdleIntervalNs() {
+ long value = TimeUnit.NANOSECONDS.convert(
+ getInt(MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG).longValue(),
+ TimeUnit.MILLISECONDS);
+ return value > 0 ? OptionalLong.of(value) : OptionalLong.empty();
+ }
+
+ // ********* Rack Configuration **********
+
+ public Optional<String> replicaSelectorClassName() {
+ return Optional.ofNullable(getString(ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG));
+ }
+
+ // ********* Replication Configuration **********
+
+ public long replicaLagTimeMaxMs() {
+ return getLong(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG);
+ }
+
+ public int replicaSocketTimeoutMs() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG);
+ }
+
+ public int replicaSocketReceiveBufferBytes() {
+ return getInt(ReplicationConfigs.REPLICA_SOCKET_RECEIVE_BUFFER_BYTES_CONFIG);
+ }
+
+ public int replicaFetchMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchWaitMaxMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG);
+ }
+
+ public int replicaFetchMinBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_MIN_BYTES_CONFIG);
+ }
+
+ public int replicaFetchResponseMaxBytes() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_CONFIG);
+ }
+
+ public int replicaFetchBackoffMs() {
+ return getInt(ReplicationConfigs.REPLICA_FETCH_BACKOFF_MS_CONFIG);
+ }
+
+ public long replicaHighWatermarkCheckpointIntervalMs() {
+ return getLong(ReplicationConfigs.REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL_MS_CONFIG);
+ }
+
+ public int fetchPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int producerPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.PRODUCER_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public int deleteRecordsPurgatoryPurgeIntervalRequests() {
+ return getInt(ReplicationConfigs.DELETE_RECORDS_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ }
+
+ public boolean autoLeaderRebalanceEnable() {
+ return getBoolean(ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG);
+ }
+
+ public long leaderImbalanceCheckIntervalSeconds() {
+ return getLong(ReplicationConfigs.LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS_CONFIG);
+ }
+
+ public long uncleanLeaderElectionCheckIntervalMs() {
+ return getLong(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG);
+ }
+
+ public Boolean uncleanLeaderElectionEnable() {
+ return getBoolean(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
+ }
+
+ public Boolean followerFetchLastTieredOffsetEnable() {
+ return getBoolean(ReplicationConfigs.FOLLOWER_FETCH_LAST_TIERED_OFFSET_ENABLE_CONFIG);
+ }
+
+ // ********* Controlled Shutdown Configuration **********
+
+ public boolean controlledShutdownEnable() {
+ return getBoolean(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG);
+ }
+
+ // ********* Group Coordinator Configuration **********
+
+ @SuppressWarnings("removal")
+ public Set<GroupType> groupCoordinatorRebalanceProtocols() {
+ return getList(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ @SuppressWarnings("removal")
+ protected void validateGroupCoordinatorRebalanceProtocols(boolean doLog) {
+ Set<GroupType> protocols = groupCoordinatorRebalanceProtocols();
+
+ if (!protocols.contains(GroupType.CLASSIC)) {
+ throw new ConfigException("Disabling the '" + GroupType.CLASSIC + "' protocol is not supported.");
+ }
+
+ if (doLog && protocols.contains(GroupType.SHARE)) {
+ log.warn("'{}' in `{}` is deprecated. " +
+ "Share groups are controlled by the 'share.version' feature. " +
+ "This config will be removed in Kafka 5.0.",
+ GroupType.SHARE,
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+
+ if (doLog && originals().containsKey(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG)) {
+ Set<GroupType> defaultProtocols = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT
+ .stream()
+ .map(s -> GroupType.valueOf(s.toUpperCase(Locale.ROOT)))
+ .collect(Collectors.toUnmodifiableSet());
+
+ Set<GroupType> missingProtocols = new HashSet<>(defaultProtocols);
+ missingProtocols.removeAll(protocols);
+
+ if (!missingProtocols.isEmpty()) {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "The following protocol(s) are currently disabled: {}. " +
+ "In Kafka 5.0, all protocols will always be enabled and controlled solely by feature versions " +
+ "(group.version, streams.version, share.version) via kafka-features.sh. " +
+ "Please remove the configuration, which will restore all protocols to the default enabled state, to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG,
+ missingProtocols.stream().map(GroupType::toString).collect(Collectors.joining(", ")));
+ } else {
+ log.warn("The config `{}` is deprecated and will be removed in Kafka 5.0. " +
+ "Please remove the configuration to prepare for the upgrade.",
+ GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG);
+ }
+ }
+ }
+
+ // ********* Metric Configuration **********
+
+ public int metricNumSamples() {
+ return getInt(MetricConfigs.METRIC_NUM_SAMPLES_CONFIG);
+ }
+
+ public long metricSampleWindowMs() {
+ return getLong(MetricConfigs.METRIC_SAMPLE_WINDOW_MS_CONFIG);
+ }
+
+ public String metricRecordingLevel() {
+ return getString(MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG);
+ }
+
+ // ********* Kafka Client Telemetry Metrics Configuration **********
+
+ public int clientTelemetryMaxBytes() {
+ return getInt(MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_CONFIG);
+ }
+
+ // ********* SSL/SASL Configuration **********
+ // Security configs may be overridden for listeners, so it is not safe to use the base values.
+ // Hence the base SSL/SASL configs are not fields of KafkaConfig, listener configs should be
+ // retrieved using KafkaConfig#valuesWithPrefixOverride
+
+ @SuppressWarnings("unchecked")
+ protected Set<String> saslEnabledMechanisms(ListenerName listenerName) {
+ Object value = valuesWithPrefixOverride(listenerName.configPrefix())
+ .get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
+ if (value != null) {
+ return Set.copyOf((List<String>) value);
+ }
+ return Set.of();
+ }
+
+ public String saslMechanismInterBrokerProtocol() {
+ return getString(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG);
+ }
+
+ // ********* Fetch Configuration **********
+
+ public int maxIncrementalFetchSessionCacheSlots() {
+ return getInt(ServerConfigs.MAX_INCREMENTAL_FETCH_SESSION_CACHE_SLOTS_CONFIG);
+ }
+
+ public int fetchMaxBytes() {
+ return getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG);
+ }
+
+ // ********* Request Limit Configuration **********
+
+ public int maxRequestPartitionSizeLimit() {
+ return getInt(ServerConfigs.MAX_REQUEST_PARTITION_SIZE_LIMIT_CONFIG);
+ }
+
+ public boolean deleteTopicEnable() {
+ return getBoolean(ServerConfigs.DELETE_TOPIC_ENABLE_CONFIG);
+ }
+
+ public String compressionType() {
+ return getString(ServerConfigs.COMPRESSION_TYPE_CONFIG);
+ }
+
+ public Integer gzipCompressionLevel() {
+ return getInt(ServerConfigs.COMPRESSION_GZIP_LEVEL_CONFIG);
+ }
+
+ public Integer lz4CompressionLevel() {
+ return getInt(ServerConfigs.COMPRESSION_LZ4_LEVEL_CONFIG);
+ }
+
+ public Integer zstdCompressionLevel() {
+ return getInt(ServerConfigs.COMPRESSION_ZSTD_LEVEL_CONFIG);
+ }
+
+ // ********* Internal Configurations **********
+
+ public boolean unstableApiVersionsEnabled() {
+ return getBoolean(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG);
+ }
+
+ public boolean unstableFeatureVersionsEnabled() {
+ return getBoolean(ServerConfigs.UNSTABLE_FEATURE_VERSIONS_ENABLE_CONFIG);
+ }
+
}