Skip to content

KAFKA-17431: Support invalid static configs for KRaft so long as dynamic configs are valid #18949

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cmccabe merged 8 commits into from
Mar 18, 2025
Merged

KAFKA-17431: Support invalid static configs for KRaft so long as dynamic configs are valid #18949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5e47e11
support invalid static configs on brokerServer if last snapshots' dyn…
kevin-wu24 Feb 14, 2025
caee8a5
code review
kevin-wu24 Feb 26, 2025
00db22f
fixing build
kevin-wu24 Feb 26, 2025
590f50e
removing magic number
kevin-wu24 Feb 26, 2025
645bcb1
code review + cleanup
kevin-wu24 Mar 4, 2025
4b81603
merging trunk
kevin-wu24 Mar 11, 2025
30e777c
handling null config record values in snapshot dynamic config load
kevin-wu24 Mar 17, 2025
6cd3e56
instantiate quotaManagers before reading dynamic config from snapshot…
kevin-wu24 Mar 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion core/src/main/scala/kafka/raft/RaftManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
}
5 changes: 3 additions & 2 deletions core/src/main/scala/kafka/server/BrokerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@cmccabe cmccabe Mar 4, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will be too late to address some of the use-cases we're concerned about, right?

It would be better to load all the config key/value pairs from the snapshot and overwrite the static configs with them, before the first time we create a KafkaConfig object.

@kevin-wu24 kevin-wu24 Mar 4, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will be too late to address some of the use-cases we're concerned about, right?

Could you illuminate some of these cases for me? It seems like we're doing this early enough in brokerServer's startup to me. Are we instead concerned that bad static configs will prevent even SharedServer#startup, which sets up KRaft and metadata publishers, from completing successfully? I guess I'm just confused as to how node can even get into a state where the static configs would crash SharedServer#startup, but somehow also have valid dynamic configs?

I'm a bit confused as to what you're proposing. It sounds like this loading config key/value pairs from the snapshot should occur before we construct the KafkaConfig in SharedServer's initialization (i.e. before we call SharedServer#start which initializes raftManager)? If so, that means we can't use raftManager to actually perform the snapshot read, right?

@cmccabe cmccabe Mar 4, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you illuminate some of these cases for me?

One use case is if we have changed the network configuration in such a way that the old static configuration isn't valid. The SSL keystore file being moved is a good example. In that case we would not want to start up with the old static configuration

If so, that means we can't use raftManager to actually perform the snapshot read, right?

It shouldn't be necessary to use raftManager to read the snapshot, since the snapshot is just a file. Maybe we could have some static utility method which finds the last snapshot file (it's just a matter of finding the one that sorts last in the folder...)

@kevin-wu24 kevin-wu24 Mar 4, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The test I wrote does "move" the SSL keystore file. By invalidating the static ssl configs on one of the nodes after shutting it down, we can successfully complete brokerServer#startup() with my changes. Without them, we fail during startup.

@cmccabe cmccabe Mar 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we want full parity with how ZK worked, we need to load the dynamic configurations prior to sending out the initial dynamic configurations. In ZK mode we actually fetched the broker configuration from ZK prior to doing this. It's possible we could do this in a different way but I'm not confident that it will solve all the possible cases.

@kevin-wu24 kevin-wu24 Mar 7, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we need to load the dynamic configurations prior to sending out the initial dynamic configurations

I'm a bit confused about what this means. Where in sharedServer do we "send out initial dynamic configurations"?

From a high level, I think the current implementation is doing the equivalent in KRaft, since the KRaft layer, which reads the snapshot, on the broker is acting like ZK (although possibly stale) to store broker configs (correct me if I'm wrong, I also will look more into this tmrw).

The only config I see that is used in sharedServer/raftManager that is also part of DynamicBrokerConfig#AllDynamicConfigs is the LISTENER_SECURITY_PROTOCOL_MAP_CONFIG. This config is read when building the KRaft network client for the broker during startup, but only the entry for the controller listener is used. From my understanding, the only way to invalidate this static config if it was previously valid would be to mess with server.properties directly and then restart the broker. Is that also a case we also want to cover with this change?

@cmccabe cmccabe Mar 17, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm a bit confused about what this means. Where in sharedServer do we "send out initial dynamic configurations"?

SharedServer doesn't. This happens from BrokerServer or ControllerServer.

From a high level, I think the current implementation is doing the equivalent in KRaft, since the KRaft layer, which reads the snapshot, on the broker is acting like ZK (although possibly stale) to store broker configs (correct me if I'm wrong, I also will look more into this tmrw).

If you are confident that this will handle the case that originally motivated this JIRA then I'm OK with doing this for now. As you recall, the case was the broker SSL keystore file being statically set to a path that no longer existed.

The only config I see that is used in sharedServer/raftManager that is also part of DynamicBrokerConfig#AllDynamicConfigs is the LISTENER_SECURITY_PROTOCOL_MAP_CONFIG. This config is read when building the KRaft network client for the broker during startup, but only the entry for the controller listener is used. From my understanding, the only way to invalidate this static config if it was previously valid would be to mess with server.properties directly and then restart the broker. Is that also a case we also want to cover with this change?

No. That configuration is static and cannot be dynamically changed.

@kevin-wu24 kevin-wu24 Mar 17, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you are confident that this will handle the case that originally motivated this JIRA then I'm OK with doing this for now. As you recall, the case was the broker SSL keystore file being statically set to a path that no longer existed.

The test I wrote shows that the implementation changes address this case, since it does the following:

  • updating the dynamic configs with the current valid static configs, which includes the keystore file
  • making sure the broker's most recent snapshot has that config update
  • shutdown the broker
  • setting the static keystore location config to an invalid file path
  • verify we can start the broker again (with my change this test passes, and without it the test throws a NoSuchFileException for the invalid file path)


/* start scheduler */
kafkaScheduler = new KafkaScheduler(config.backgroundThreads)
Expand All @@ -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())
Expand Down
55 changes: 53 additions & 2 deletions core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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)

@chia7712 chia7712 Aug 24, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dynamicPerBrokerConfigs should be replaced by dynamicDefaultConfigs. I have opened https://issues.apache.org/jira/browse/KAFKA-19642 to fix it

configHandler.processConfigChanges(config.brokerId.toString, dynamicPerBrokerConfigs)
})
})
}
}

class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging {
Expand Down
86 changes: 61 additions & 25 deletions core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand All @@ -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())
Expand Down Expand Up @@ -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))
Expand Down