diff --git a/core/src/main/scala/kafka/raft/RaftManager.scala b/core/src/main/scala/kafka/raft/RaftManager.scala index 9ff75b126ad56..e55e91eaaef70 100644 --- a/core/src/main/scala/kafka/raft/RaftManager.scala +++ b/core/src/main/scala/kafka/raft/RaftManager.scala @@ -104,13 +104,15 @@ trait RaftManager[T] { def replicatedLog: ReplicatedLog def voterNode(id: Int, listener: ListenerName): Option[Node] + + def recordSerde: RecordSerde[T] } class KafkaRaftManager[T]( clusterId: String, config: KafkaConfig, metadataLogDirUuid: Uuid, - recordSerde: RecordSerde[T], + serde: RecordSerde[T], topicPartition: TopicPartition, topicId: Uuid, time: Time, @@ -298,4 +300,6 @@ class KafkaRaftManager[T]( override def voterNode(id: Int, listener: ListenerName): Option[Node] = { client.voterNode(id, listener).toScala } + + override def recordSerde: RecordSerde[T] = serde } diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala index 5a65e0288779a..60ed42e0976c9 100644 --- a/core/src/main/scala/kafka/server/BrokerServer.scala +++ b/core/src/main/scala/kafka/server/BrokerServer.scala @@ -194,7 +194,10 @@ class BrokerServer( info("Starting broker") val clientMetricsReceiverPlugin = new ClientMetricsReceiverPlugin() + config.dynamicConfig.initialize(Some(clientMetricsReceiverPlugin)) + quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-") + DynamicBrokerConfig.readDynamicBrokerConfigsFromSnapshot(raftManager, config, quotaManagers) /* start scheduler */ kafkaScheduler = new KafkaScheduler(config.backgroundThreads) @@ -203,8 +206,6 @@ class BrokerServer( /* register broker metrics */ brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.isRemoteStorageSystemEnabled()) - quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-") - logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size) metadataCache = MetadataCache.kRaftMetadataCache(config.nodeId, () => raftManager.client.kraftVersion()) diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala index 33b7192c36d2b..b9b25c1afebaa 100755 --- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala +++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala @@ -23,23 +23,28 @@ import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.locks.ReentrantReadWriteLock import kafka.log.{LogCleaner, LogManager} import kafka.network.{DataPlaneAcceptor, SocketServer} +import kafka.raft.KafkaRaftManager import kafka.server.DynamicBrokerConfig._ import kafka.utils.{CoreUtils, Logging} import org.apache.kafka.common.Reconfigurable import org.apache.kafka.network.EndPoint import org.apache.kafka.common.config.internals.BrokerSecurityConfigs -import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, SaslConfigs, SslConfigs} +import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource, SaslConfigs, SslConfigs} +import org.apache.kafka.common.metadata.{ConfigRecord, MetadataRecordType} import org.apache.kafka.common.metrics.{Metrics, MetricsReporter} import org.apache.kafka.common.network.{ListenerName, ListenerReconfigurable} import org.apache.kafka.common.security.authenticator.LoginManager -import org.apache.kafka.common.utils.{ConfigUtils, Utils} +import org.apache.kafka.common.utils.{BufferSupplier, ConfigUtils, Utils} import org.apache.kafka.coordinator.transaction.TransactionLogConfig import org.apache.kafka.network.SocketServerConfigs +import org.apache.kafka.raft.KafkaRaftClient import org.apache.kafka.server.{ProcessRole, DynamicThreadPool} +import org.apache.kafka.server.common.ApiMessageAndVersion import org.apache.kafka.server.config.{ServerConfigs, ServerLogConfigs, ServerTopicConfigSynonyms} import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig import org.apache.kafka.server.metrics.{ClientMetricsReceiverPlugin, MetricConfigs} import org.apache.kafka.server.telemetry.ClientTelemetry +import org.apache.kafka.snapshot.RecordsSnapshotReader import org.apache.kafka.storage.internals.log.{LogConfig, ProducerStateManagerConfig} import scala.collection._ @@ -185,6 +190,52 @@ object DynamicBrokerConfig { } props } + + private[server] def readDynamicBrokerConfigsFromSnapshot( + raftManager: KafkaRaftManager[ApiMessageAndVersion], + config: KafkaConfig, + quotaManagers: QuotaFactory.QuotaManagers + ): Unit = { + def putOrRemoveIfNull(props: Properties, key: String, value: String): Unit = { + if (value == null) { + props.remove(key) + } else { + props.put(key, value) + } + } + raftManager.replicatedLog.latestSnapshotId().ifPresent(latestSnapshotId => { + raftManager.replicatedLog.readSnapshot(latestSnapshotId).ifPresent(rawSnapshotReader => { + val reader = RecordsSnapshotReader.of( + rawSnapshotReader, + raftManager.recordSerde, + BufferSupplier.create(), + KafkaRaftClient.MAX_BATCH_SIZE_BYTES, + true + ) + val dynamicPerBrokerConfigs = new Properties() + val dynamicDefaultConfigs = new Properties() + while (reader.hasNext) { + val batch = reader.next() + batch.forEach(record => { + if (record.message().apiKey() == MetadataRecordType.CONFIG_RECORD.id) { + val configRecord = record.message().asInstanceOf[ConfigRecord] + if (DynamicBrokerConfig.AllDynamicConfigs.contains(configRecord.name()) && + configRecord.resourceType() == ConfigResource.Type.BROKER.id()) { + if (configRecord.resourceName().isEmpty) { + putOrRemoveIfNull(dynamicDefaultConfigs, configRecord.name(), configRecord.value()) + } else if (configRecord.resourceName() == config.brokerId.toString) { + putOrRemoveIfNull(dynamicPerBrokerConfigs, configRecord.name(), configRecord.value()) + } + } + } + }) + } + val configHandler = new BrokerConfigHandler(config, quotaManagers) + configHandler.processConfigChanges("", dynamicPerBrokerConfigs) + configHandler.processConfigChanges(config.brokerId.toString, dynamicPerBrokerConfigs) + }) + }) + } } class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging { diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index 8b9fa754a2a93..34d1cf9b1c7ca 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -57,7 +57,7 @@ import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer} import org.apache.kafka.coordinator.transaction.TransactionLogConfig import org.apache.kafka.network.SocketServerConfigs -import org.apache.kafka.server.config.{ReplicationConfigs, ServerConfigs, ServerLogConfigs, ServerTopicConfigSynonyms} +import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs, ServerTopicConfigSynonyms} import org.apache.kafka.server.metrics.{KafkaYammerMetrics, MetricConfigs} import org.apache.kafka.server.record.BrokerCompressionType import org.apache.kafka.server.util.ShutdownableThread @@ -117,30 +117,7 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup clearLeftOverProcessorMetrics() // clear metrics left over from other tests so that new ones can be tested (0 until numServers).foreach { brokerId => - - val props = TestUtils.createBrokerConfig(brokerId) - props.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, s"$SecureInternal://localhost:0, $SecureExternal://localhost:0") - props ++= securityProps(sslProperties1, TRUSTSTORE_PROPS) - // Ensure that we can support multiple listeners per security protocol and multiple security protocols - props.put(SocketServerConfigs.LISTENERS_CONFIG, s"$SecureInternal://localhost:0, $SecureExternal://localhost:0") - props.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, s"PLAINTEXT:PLAINTEXT, $SecureInternal:SSL, $SecureExternal:SASL_SSL, CONTROLLER:$controllerListenerSecurityProtocol") - props.put(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG, SecureInternal) - props.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "requested") - props.put(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN") - props.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, kafkaServerSaslMechanisms.mkString(",")) - props.put(ServerLogConfigs.LOG_SEGMENT_BYTES_CONFIG, "1048576") // low value to test log rolling on config update - props.put(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG, "2") // greater than one to test reducing threads - props.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, 1680000000.toString) - props.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, 168.toString) - - props ++= sslProperties1 - props ++= securityProps(sslProperties1, KEYSTORE_PROPS, listenerPrefix(SecureInternal)) - - // Set invalid top-level properties to ensure that listener config is used - // Don't set any dynamic configs here since they get overridden in tests - props ++= invalidSslProperties - props ++= securityProps(invalidSslProperties, KEYSTORE_PROPS) - props ++= securityProps(sslProperties1, KEYSTORE_PROPS, listenerPrefix(SecureExternal)) + val props = defaultStaticConfig(brokerId) val kafkaConfig = KafkaConfig.fromProps(props) @@ -158,6 +135,33 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup TestMetricsReporter.testReporters.clear() } + def defaultStaticConfig(brokerId: Int): Properties = { + val props = TestUtils.createBrokerConfig(brokerId) + props.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, s"$SecureInternal://localhost:0, $SecureExternal://localhost:0") + props ++= securityProps(sslProperties1, TRUSTSTORE_PROPS) + // Ensure that we can support multiple listeners per security protocol and multiple security protocols + props.put(SocketServerConfigs.LISTENERS_CONFIG, s"$SecureInternal://localhost:0, $SecureExternal://localhost:0") + props.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, s"PLAINTEXT:PLAINTEXT, $SecureInternal:SSL, $SecureExternal:SASL_SSL, CONTROLLER:$controllerListenerSecurityProtocol") + props.put(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG, SecureInternal) + props.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "requested") + props.put(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN") + props.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, kafkaServerSaslMechanisms.mkString(",")) + props.put(ServerLogConfigs.LOG_SEGMENT_BYTES_CONFIG, "1048576") // low value to test log rolling on config update + props.put(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG, "2") // greater than one to test reducing threads + props.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, 1680000000.toString) + props.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, 168.toString) + + props ++= sslProperties1 + props ++= securityProps(sslProperties1, KEYSTORE_PROPS, listenerPrefix(SecureInternal)) + + // Set invalid top-level properties to ensure that listener config is used + // Don't set any dynamic configs here since they get overridden in tests + props ++= invalidSslProperties + props ++= securityProps(invalidSslProperties, KEYSTORE_PROPS) + props ++= securityProps(sslProperties1, KEYSTORE_PROPS, listenerPrefix(SecureExternal)) + props + } + @AfterEach override def tearDown(): Unit = { clientThreads.foreach(_.interrupt()) @@ -1091,6 +1095,38 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup verifyConfiguration(true) } + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames) + @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll")) + def testServersCanStartWithInvalidStaticConfigsAndValidDynamicConfigs(quorum: String, groupProtocol: String): Unit = { + // modify snapshot interval config to explicitly take snapshot on a broker with valid dynamic configs + val props = defaultStaticConfig(numServers) + props.put(KRaftConfigs.METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG, "10000") + + val kafkaConfig = KafkaConfig.fromProps(props) + val newBroker = createBroker(kafkaConfig).asInstanceOf[BrokerServer] + servers += newBroker + + alterSslKeystoreUsingConfigCommand(sslProperties1, listenerPrefix(SecureExternal)) + + TestUtils.ensureConsistentKRaftMetadata(servers, controllerServer) + + TestUtils.waitUntilTrue( + () => newBroker.raftManager.replicatedLog.latestSnapshotId().isPresent, + "metadata snapshot not present on broker", + 30000L + ) + + // shutdown broker and attempt to restart it after invalidating its static configurations + newBroker.shutdown() + newBroker.awaitShutdown() + + val invalidStaticConfigs = defaultStaticConfig(newBroker.config.brokerId) + invalidStaticConfigs.putAll(securityProps(invalidSslConfigs, KEYSTORE_PROPS, listenerPrefix(SecureExternal))) + newBroker.config.updateCurrentConfig(KafkaConfig.fromProps(invalidStaticConfigs)) + + newBroker.startup() + } + private def awaitInitialPositions(consumer: Consumer[_, _]): Unit = { TestUtils.pollUntilTrue(consumer, () => !consumer.assignment.isEmpty, "Timed out while waiting for assignment") consumer.assignment.forEach(tp => consumer.position(tp))