Skip to content

KAFKA-15295: Add config validation when remote storage is enabled on a topic #14176

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
divijvaidya merged 9 commits into from
Aug 15, 2023
Merged

KAFKA-15295: Add config validation when remote storage is enabled on a topic #14176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c56142f
KAFKA-15295: Add config validation when remote storage is enabled on …
kamalcph Aug 9, 2023
e184917
Fix the test failures.
kamalcph Aug 13, 2023
1ecca4e
update the comments for clarity
kamalcph Aug 13, 2023
4f15a3a
Addressed the review comments.
kamalcph Aug 14, 2023
d979f01
Addressed the review-2 comments.
kamalcph Aug 14, 2023
21bf7a3
Addressed the review comments.
kamalcph Aug 15, 2023
d22d487
reduce the method access specifier to private
kamalcph Aug 15, 2023
93ac9db
Fix the unit tests.
kamalcph Aug 15, 2023
0b02178
Retry the `getLog` operation until the timeout.
kamalcph Aug 15, 2023
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
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/log/LogManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1376,7 +1376,7 @@ object LogManager {
keepPartitionMetadataFile: Boolean): LogManager = {
val defaultProps = config.extractLogConfigMap

LogConfig.validateValues(defaultProps)
LogConfig.validateBrokerLogConfigValues(defaultProps, config.isRemoteLogStorageSystemEnabled)
val defaultLogConfig = new LogConfig(defaultProps)

val cleanerConfig = LogCleaner.cleanerConfig(config)
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ import scala.collection.mutable
* in the same RPC, BROKER_LOGGER is not really a dynamic configuration in the same sense
* as the others. It is not persisted to the metadata log (or to ZK, when we're in that mode).
*/
class ControllerConfigurationValidator extends ConfigurationValidator {
class ControllerConfigurationValidator(kafkaConfig: KafkaConfig) extends ConfigurationValidator {
private def validateTopicName(
name: String
): Unit = {
Expand Down Expand Up @@ -106,7 +106,7 @@ class ControllerConfigurationValidator extends ConfigurationValidator {
throw new InvalidConfigurationException("Null value not supported for topic configs: " +
nullTopicConfigs.mkString(","))
}
LogConfig.validate(properties)
LogConfig.validate(properties, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
case BROKER => validateBrokerName(resource.name())
case _ => throwExceptionForUnknownResourceType(resource)
}
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/ControllerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ class ControllerServer(
setMetrics(quorumControllerMetrics).
setCreateTopicPolicy(createTopicPolicy.asJava).
setAlterConfigPolicy(alterConfigPolicy.asJava).
setConfigurationValidator(new ControllerConfigurationValidator()).
setConfigurationValidator(new ControllerConfigurationValidator(sharedServer.brokerConfig)).

@kamalcph kamalcph Aug 10, 2023 •
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.

what is the difference between using sharedServer.brokerConfig vs sharedServer.controllerConfig? Seems to be the same.

@divijvaidya divijvaidya Aug 11, 2023 •
edited
Loading

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.

sharedServer represents the case when kraft controller is run on a broker and not as an independent node. In such case a broker has two responsibility, one to act as a controller and another to act as a broker. These two configs represent the configurations associated with node's role as a broker and as a controller.

In our case here, when createTopic or alterConfig is called to enable TS for a topic, it will be forwarded to the controller. Controller will validate the config using ControllerConfigurationValidator before applying it. The assumption here is that the broker level configuration has already been validated before forwarding. Now, this is the first case where we want to make an assertion consisting of both broker level configuration and topic level configuration. I would ideally have wanted to fail fast with this at broker itself before it is sent to controller by adding the validation at

kafka/core/src/main/scala/kafka/server/ConfigAdminManager.scala

Line 155 in f137da0

// Nothing to do.

@divijvaidya divijvaidya Aug 11, 2023

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.

Thinking more about it...this is more complicated than I expected.

We want to block enablement of a Topic level config if all broker don't have TS enabled on them. We need a way to determine that the TS has been enabled on all brokers. In Kraft world, no component has a view of all broker configs, not even the controller (correct me if I am wrong here) because broker level config is in their separate server.properties files.

As an example, what happens when we are in a rolling restart, some brokers have TS enabled on them and some don't. We send an alter config call to enable TS for a topic, it hits the one which has TS enabled, this broker forwards it to the controller and controller will send the config update to all brokers. When another broker which doesn't have TS enabled gets this config change, it "should" fail to apply it. But failing now is too late since alterConfig has already succeeded since controller->broker config propagation is done async.

With this limitation in mind, the ideal solution is:

  1. add a logic in controller such that it knows broker level config of all brokers (does it already know that in metadata?)
  2. when request to enable TS for a topic arrives, ensure that all brokers have TS enabled, if not, then reject.

@kamalcph kamalcph Aug 12, 2023

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.

Can we do this validation separately? Or, as part of KAFKA-15267 ticket. cc @clolov.

@divijvaidya divijvaidya Aug 14, 2023

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.

I created a JIRA for the scenario mentioned above. We can consider it separately. https://issues.apache.org/jira/browse/KAFKA-15341

@showuon showuon Aug 14, 2023

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.

I think the scenario from @divijvaidya is too complicated. I don't think we have any other similar config validations like this (from broker 1 has different config with broker 2). IMO, this PR already adds validation for it, and for the edge case, we can still fail the request with clear logs, it should be good enough. WDYT?

@divijvaidya divijvaidya Aug 14, 2023

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.

Nope, I would disagree that the scenario is complicated. For larger clusters containing hundreds of nodes, rolling restart can take a long time. Any functionality that we introduce in Kafka code base should be able to handle scenarios where some brokers have features enabled and others don't. In existing code base this is achieved by using the "features" [1]. When a broker sends metadata to the controller, it will also send "features" that it supports. In our situation, we need to add TS as a "feature". So during rolling restart, controller knows that not all brokers have the correct feature and will reject any call to enable TS for a topic. After rolling restart is complete, controller will know that all brokers have TS feature on them, hence, it can start enabling TS for topic.

[1]

kafka/core/src/main/scala/kafka/server/BrokerFeatures.scala

Line 33 in 43751d8

class BrokerFeatures private (@volatile var supportedFeatures: Features[SupportedVersionRange]) {

@divijvaidya divijvaidya Aug 14, 2023

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.

Nevertheless, we can discuss it in separate JIRA that I created above.

setStaticConfig(config.originals).
setBootstrapMetadata(bootstrapMetadata).
setFatalFaultHandler(sharedServer.fatalQuorumControllerFaultHandler).
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/ZkAdminManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ class ZkAdminManager(val config: KafkaConfig,
this.logIdent = "[Admin Manager on Broker " + config.brokerId + "]: "

private val topicPurgatory = DelayedOperationPurgatory[DelayedOperation]("topic", config.brokerId)
private val adminZkClient = new AdminZkClient(zkClient)
private val adminZkClient = new AdminZkClient(zkClient, Some(config))
private val configHelper = new ConfigHelper(metadataCache, config, new ZkConfigRepository(adminZkClient))

private val createTopicPolicy =
Expand Down
15 changes: 10 additions & 5 deletions core/src/main/scala/kafka/zk/AdminZkClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,11 @@
*/
package kafka.zk

import java.util.{Optional, Properties}
import java.util.{Collections, Optional, Properties}
import kafka.admin.RackAwareMode
import kafka.common.TopicAlreadyMarkedForDeletionException
import kafka.controller.ReplicaAssignment
import kafka.server.{ConfigEntityName, ConfigType, DynamicConfig}
import kafka.server.{ConfigEntityName, ConfigType, DynamicConfig, KafkaConfig}
import kafka.utils._
import kafka.utils.Implicits._
import org.apache.kafka.admin.{AdminUtils, BrokerMetadata}
Expand All @@ -40,7 +40,8 @@ import scala.collection.{Map, Seq}
* This is an internal class and no compatibility guarantees are provided,
* see org.apache.kafka.clients.admin.AdminClient for publicly supported APIs.
*/
class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
class AdminZkClient(zkClient: KafkaZkClient,
kafkaConfig: Option[KafkaConfig] = None) extends Logging {

/**
* Creates the topic with given configuration
Expand Down Expand Up @@ -159,7 +160,9 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
partitionReplicaAssignment.keys.filter(_ >= 0).sum != sequenceSum)
throw new InvalidReplicaAssignmentException("partitions should be a consecutive 0-based integer sequence")

LogConfig.validate(config)
LogConfig.validate(config,
kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()),
kafkaConfig.exists(_.isRemoteLogStorageSystemEnabled))
}

private def writeTopicPartitionAssignment(topic: String, replicaAssignment: Map[Int, ReplicaAssignment],
Expand Down Expand Up @@ -475,7 +478,9 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
if (!zkClient.topicExists(topic))
throw new UnknownTopicOrPartitionException(s"Topic '$topic' does not exist.")
// remove the topic overrides
LogConfig.validate(configs)
LogConfig.validate(configs,
kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()),
kafkaConfig.exists(_.isRemoteLogStorageSystemEnabled))
}

/**
Expand Down
Loading