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/14 19:40:10 UTC
(kafka) branch trunk updated: KAFKA-21222 Add broker-level configs (part 2) (#23797)
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 27801d794b4 KAFKA-21222 Add broker-level configs (part 2) (#23797)
27801d794b4 is described below
commit 27801d794b4bcb3fad6d2cff3a3d2137ec779174
Author: Federico Valeri <fe...@gmail.com>
AuthorDate: Wed Oct 14 21:40:02 2026 +0200
KAFKA-21222 Add broker-level configs (part 2) (#23797)
Follow-up to #23729: move DynamicClusterMirrorConfig to the server
module as Java code, taking a Supplier<AbstractKafkaConfig> instead of
KafkaBroker, and drop the redundant check on
mirror.metadata.refresh.interval.ms already enforced by ConfigDef.
Reviewers: Gaurav Narula <ga...@apple.com>, Luke Chen
<sh...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../scala/kafka/server/DynamicBrokerConfig.scala | 32 +---------
.../kafka/server/DynamicBrokerConfigTest.scala | 74 +++-------------------
.../kafka/server/config/DynamicBrokerConfig.java | 39 ++++++++++--
.../server/config/DynamicBrokerConfigTest.java | 39 ++++++++++++
4 files changed, 87 insertions(+), 97 deletions(-)
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 51ec0ac9f6a..43db686f51d 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -39,8 +39,8 @@ 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.coordinator.mirror.ClusterMirrorConfigs
-import org.apache.kafka.server.config.{BrokerReconfigurable => JBrokerReconfigurable, DynamicConfig, DynamicProducerStateManagerConfig, ServerConfigs, ServerLogConfigs, DynamicBrokerConfig => JDynamicBrokerConfig}
+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.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.metrics.{ClientTelemetryExporterPlugin, MetricConfigs}
import org.apache.kafka.server.quota.QuotaFactory
@@ -202,7 +202,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
addBrokerReconfigurable(kafkaServer.socketServer)
addBrokerReconfigurable(new DynamicProducerStateManagerConfig(kafkaServer.logManager.producerStateManagerConfig))
addBrokerReconfigurable(new DynamicRemoteLogConfig(kafkaServer))
- addBrokerReconfigurable(new DynamicClusterMirrorConfig(kafkaServer))
+ addBrokerReconfigurable(new DynamicClusterMirrorConfig(() => kafkaServer.config))
}
/**
@@ -1000,29 +1000,3 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
}
}
}
-
-class DynamicClusterMirrorConfig(server: KafkaBroker) extends BrokerReconfigurable with Logging {
- override def reconfigurableConfigs: util.Set[String] = {
- JDynamicBrokerConfig.DynamicClusterMirrorConfig.RECONFIGURABLE_CONFIGS
- }
-
- override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
- val newFetchers = newConfig.getInt(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG)
- val oldFetchers = server.config.getInt(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG)
- if (newFetchers != oldFetchers) {
- val errorMsg = s"Dynamic thread count update validation failed for ${ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG}=$newFetchers"
- if (newFetchers < oldFetchers / 2)
- throw new ConfigException(s"$errorMsg, Value should be at least half the current value $oldFetchers")
- if (newFetchers > oldFetchers * 2)
- throw new ConfigException(s"$errorMsg, Value should not be greater than double the current value $oldFetchers")
- }
-
- val newInterval = newConfig.getLong(ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG)
- if (newInterval < 0)
- throw new ConfigException(s"${ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG} must be >= 0")
- }
-
- override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
- // TODO: Implement reconfiguration when MirrorMetadataManager is available (KAFKA-21233)
- }
-}
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index d44b3f6cdf9..ee721a4151c 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -17,43 +17,42 @@
package kafka.server
-import java.{lang, util}
-import java.util.{Optional, Properties, Map => JMap}
-import java.util.concurrent.{CompletionStage, TimeUnit}
-import java.util.concurrent.atomic.AtomicReference
import kafka.network.{DataPlaneAcceptor, SocketServer}
import kafka.utils.TestUtils
-import org.apache.kafka.common.{Endpoint, Reconfigurable, Uuid}
import org.apache.kafka.common.acl.{AclBinding, AclBindingFilter}
import org.apache.kafka.common.config.{ConfigException, SslConfigs}
import org.apache.kafka.common.internals.Plugin
import org.apache.kafka.common.metrics.{JmxReporter, KafkaMetric, Metrics, MetricsReporter}
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.{Endpoint, Reconfigurable, Uuid}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.coordinator.share.ShareCoordinatorConfig
-import org.apache.kafka.raft.{KRaftConfigs, QuorumConfig}
import org.apache.kafka.network.{SocketServerConfigs, SocketServer => JSocketServer}
+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.coordinator.mirror.ClusterMirrorConfigs
-import org.apache.kafka.server.config.{ReplicationConfigs, ServerConfigs, ServerLogConfigs, DynamicBrokerConfig => JDynamicBrokerConfig}
+import org.apache.kafka.server.config.{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
-import org.apache.kafka.server.telemetry.{ClientTelemetry, ClientTelemetryContext, ClientTelemetryExporter, ClientTelemetryExporterProvider, ClientTelemetryPayload, ClientTelemetryReceiver}
+import org.apache.kafka.server.telemetry._
import org.apache.kafka.server.util.KafkaScheduler
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogManager, ProducerStateManagerConfig}
import org.apache.kafka.test.MockMetricsReporter
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
import org.mockito.ArgumentMatchers.{anySet, anyString}
+import org.mockito.Mockito._
import org.mockito.{ArgumentCaptor, ArgumentMatchers, Mockito}
-import org.mockito.Mockito.{mock, never, times, verify, verifyNoMoreInteractions, when}
-import scala.jdk.CollectionConverters._
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{CompletionStage, TimeUnit}
+import java.util.{Optional, Properties, Map => JMap}
+import java.{lang, util}
import scala.collection.Set
+import scala.jdk.CollectionConverters._
class DynamicBrokerConfigTest {
@@ -1345,59 +1344,6 @@ class DynamicBrokerConfigTest {
assertEquals(true, config.groupCoordinatorConfig.streamsGroupAssignorOffloadEnable())
}
- @Test
- def testDynamicClusterMirrorConfigReconfigurableConfigs(): Unit = {
- assertTrue(JDynamicBrokerConfig.DynamicClusterMirrorConfig.RECONFIGURABLE_CONFIGS.contains(
- ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG))
- assertTrue(JDynamicBrokerConfig.DynamicClusterMirrorConfig.RECONFIGURABLE_CONFIGS.contains(
- ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG))
- }
-
- @Test
- def testDynamicClusterMirrorConfigValidation(): Unit = {
- val origProps = TestUtils.createBrokerConfig(0, port = 8181)
- origProps.put(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG, "4")
- origProps.put(ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG, "60000")
- val config = KafkaConfig(origProps)
- val serverMock = mock(classOf[KafkaBroker])
- when(serverMock.config).thenReturn(config)
-
- config.dynamicConfig.initialize(None)
- config.dynamicConfig.addBrokerReconfigurable(new DynamicClusterMirrorConfig(serverMock))
-
- val validProps = new Properties()
- validProps.put(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG, "6")
- config.dynamicConfig.validate(validProps, perBrokerConfig = true)
-
- val zeroFetchers = new Properties()
- zeroFetchers.put(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG, "0")
- val err1 = assertThrows(classOf[ConfigException],
- () => config.dynamicConfig.validate(zeroFetchers, perBrokerConfig = true))
- assertTrue(err1.getMessage.contains("Value must be at least 1"))
-
- val tooFewFetchers = new Properties()
- tooFewFetchers.put(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG, "1")
- val err2 = assertThrows(classOf[ConfigException],
- () => config.dynamicConfig.validate(tooFewFetchers, perBrokerConfig = true))
- assertTrue(err2.getMessage.contains("Value should be at least half the current value 4"))
-
- val tooManyFetchers = new Properties()
- tooManyFetchers.put(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG, "9")
- val err3 = assertThrows(classOf[ConfigException],
- () => config.dynamicConfig.validate(tooManyFetchers, perBrokerConfig = true))
- assertTrue(err3.getMessage.contains("Value should not be greater than double the current value 4"))
-
- val negativeInterval = new Properties()
- negativeInterval.put(ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG, "-1")
- val err4 = assertThrows(classOf[ConfigException],
- () => config.dynamicConfig.validate(negativeInterval, perBrokerConfig = true))
- assertTrue(err4.getMessage.contains(ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG))
-
- val zeroInterval = new Properties()
- zeroInterval.put(ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG, "0")
- config.dynamicConfig.validate(zeroInterval, perBrokerConfig = true)
- }
-
@Test
def testDynamicShareCoordinatorConfig(): Unit = {
assertTrue(ShareCoordinatorConfig.RECONFIGURABLE_CONFIGS.contains(ShareCoordinatorConfig.CACHED_BUFFER_MAX_BYTES_CONFIG))
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 b38f8f92e12..e5df7b6298a 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
@@ -41,6 +41,7 @@ import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.function.Function;
+import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -292,13 +293,43 @@ public class DynamicBrokerConfig {
RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP);
}
- public static class DynamicClusterMirrorConfig {
+ public static class DynamicQuotaConfig {
+ public static final Set<String> RECONFIGURABLE_CONFIGS = QuotaConfig.BROKER_QUOTA_CONFIGS;
+ }
+
+ public static class DynamicClusterMirrorConfig implements BrokerReconfigurable {
public static final Set<String> RECONFIGURABLE_CONFIGS = Set.of(
ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG,
ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG);
- }
- public static class DynamicQuotaConfig {
- public static final Set<String> RECONFIGURABLE_CONFIGS = QuotaConfig.BROKER_QUOTA_CONFIGS;
+ private final Supplier<AbstractKafkaConfig> currentConfig;
+
+ public DynamicClusterMirrorConfig(Supplier<AbstractKafkaConfig> currentConfig) {
+ this.currentConfig = currentConfig;
+ }
+
+ @Override
+ public Set<String> reconfigurableConfigs() {
+ return RECONFIGURABLE_CONFIGS;
+ }
+
+ @Override
+ public void validateReconfiguration(AbstractKafkaConfig newConfig) {
+ int newFetchers = newConfig.getInt(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG);
+ int oldFetchers = currentConfig.get().getInt(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG);
+ if (newFetchers != oldFetchers) {
+ String errorMsg = "Dynamic thread count update validation failed for "
+ + ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG + "=" + newFetchers;
+ if (newFetchers < oldFetchers / 2)
+ throw new ConfigException(errorMsg + ", Value should be at least half the current value " + oldFetchers);
+ if (newFetchers > oldFetchers * 2)
+ throw new ConfigException(errorMsg + ", Value should not be greater than double the current value " + oldFetchers);
+ }
+ }
+
+ @Override
+ public void reconfigure(AbstractKafkaConfig oldConfig, AbstractKafkaConfig newConfig) {
+ // TODO: Implement reconfiguration when MirrorMetadataManager is available (KAFKA-21233)
+ }
}
}
diff --git a/server/src/test/java/org/apache/kafka/server/config/DynamicBrokerConfigTest.java b/server/src/test/java/org/apache/kafka/server/config/DynamicBrokerConfigTest.java
index 31cbc6bd671..0dd2811fc3c 100644
--- a/server/src/test/java/org/apache/kafka/server/config/DynamicBrokerConfigTest.java
+++ b/server/src/test/java/org/apache/kafka/server/config/DynamicBrokerConfigTest.java
@@ -17,7 +17,10 @@
package org.apache.kafka.server.config;
import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.provider.FileConfigProvider;
+import org.apache.kafka.coordinator.mirror.ClusterMirrorConfigs;
+import org.apache.kafka.server.config.DynamicBrokerConfig.DynamicClusterMirrorConfig;
import org.junit.jupiter.api.Test;
@@ -29,6 +32,10 @@ import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
public class DynamicBrokerConfigTest {
@@ -79,4 +86,36 @@ public class DynamicBrokerConfigTest {
assertEquals(List.of("anything"), DynamicBrokerConfig.brokerConfigSynonyms("anything", false));
}
+
+ @Test
+ public void testDynamicClusterMirrorConfigReconfigurableConfigs() {
+ assertTrue(DynamicClusterMirrorConfig.RECONFIGURABLE_CONFIGS.contains(
+ ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG));
+ assertTrue(DynamicClusterMirrorConfig.RECONFIGURABLE_CONFIGS.contains(
+ ClusterMirrorConfigs.MIRROR_METADATA_REFRESH_INTERVAL_MS_CONFIG));
+ }
+
+ @Test
+ public void testDynamicClusterMirrorConfigValidation() {
+ AbstractKafkaConfig currentConfig = mockConfig(4);
+ DynamicClusterMirrorConfig dynamicConfig = new DynamicClusterMirrorConfig(() -> currentConfig);
+
+ assertDoesNotThrow(() -> dynamicConfig.validateReconfiguration(mockConfig(6)));
+
+ ConfigException err1 = assertThrows(ConfigException.class,
+ () -> dynamicConfig.validateReconfiguration(mockConfig(1)));
+ assertTrue(err1.getMessage().contains("Value should be at least half the current value 4"));
+
+ ConfigException err2 = assertThrows(ConfigException.class,
+ () -> dynamicConfig.validateReconfiguration(mockConfig(9)));
+ assertTrue(err2.getMessage().contains("Value should not be greater than double the current value 4"));
+
+ assertDoesNotThrow(() -> dynamicConfig.validateReconfiguration(mockConfig(4)));
+ }
+
+ private static AbstractKafkaConfig mockConfig(int fetchers) {
+ AbstractKafkaConfig config = mock(AbstractKafkaConfig.class);
+ when(config.getInt(ClusterMirrorConfigs.MIRROR_NUM_REPLICA_FETCHERS_CONFIG)).thenReturn(fetchers);
+ return config;
+ }
}