KAFKA-15295: Add config validation when remote storage is enabled on a topic#14176
Conversation
divijvaidya
left a comment
divijvaidya
left a comment
There was a problem hiding this comment.
Thank you for this change Kamal!
Could we also add integ tests (from client side) to validate the semantic behaviour? We want a test that will throw configException when it creates a topic and also when it alters existing topic. One of the possible place to add this is TopicAdminTest. Let me get back tomorrow with a better suited test suite. Please update PR if you already find a test class by that time.
I am yet to complete the review so leaving some initial comments. Things that I am yet to check (will do tomorrow):
- where does kraft create topic and alter topic config invoke the validation because we seem to only be changing the controller config validation.
There was a problem hiding this comment.
Using the constant from RemoteLogMangerConfig requires adding storage module as dependency for server-common module which brings circular dependency.
Reusing the constant can be done in the next PR.
|
Regarding my previous comment:
It does it in ControllerConfigurationValidator so we should be good with just changing it as you have done in this PR.
We need to add tests at:
In a separate ticket later, we will add tests that validate TS related configuration hierarchy and changes for cluster, broker and topic level. |
There was a problem hiding this comment.
what is the difference between using sharedServer.brokerConfig vs sharedServer.controllerConfig? Seems to be the same.
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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:
- add a logic in controller such that it knows broker level config of all brokers (does it already know that in metadata?)
- when request to enable TS for a topic arrives, ensure that all brokers have TS enabled, if not, then reject.
There was a problem hiding this comment.
Can we do this validation separately? Or, as part of KAFKA-15267 ticket. cc @clolov.
There was a problem hiding this comment.
I created a JIRA for the scenario mentioned above. We can consider it separately. https://issues.apache.org/jira/browse/KAFKA-15341
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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]
There was a problem hiding this comment.
Nevertheless, we can discuss it in separate JIRA that I created above.
|
Added the unit and integration test. Please take another look. Thanks! |
|
We have added sufficient validations while creating/updating the topic. If the user didn't override the Checking all the topic configs while changing the dynamic broker config is out of scope for this PR. |
|
Thanks for the timely review! Addressed most of your comments. Please take a look when you get chance! |
divijvaidya
left a comment
divijvaidya
left a comment
There was a problem hiding this comment.
Thank you Kamal for your patience so far. I believe this should be last round of review. We are almost there.
| new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"), | ||
| AlterConfigOp.OpType.SET)) | ||
| ) | ||
| admin.incrementalAlterConfigs(configs).all().get() |
There was a problem hiding this comment.
This validation might not be enough. As an example, if there is a bug in propagating the config to remote log manager, this config change will be a no-op on the server.
I will suggest to add:
val log = brokers.head.logManager.getLog(tp).get
TestUtils.waitUntilTrue(() => {
log.config.remoteStorageEnable()
}, s"remote storage is not enabled for log with config=${log.config}")
(same for other tests)
There was a problem hiding this comment.
Adding this validation will catch any regression. This requires extra amount of work as we may have to validate all the applied/changed configs in the topic by describing the topic configs. We cannot directly access the LogConfig as it doesn't support querying a config by name so skipping this comment.
There was a problem hiding this comment.
Addressed with latest commit.
| val admin = createAdminClient() | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| TestUtils.createTopicWithAdmin(admin, testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig) |
There was a problem hiding this comment.
Could be replaced with:
createTopic(testTopicName, numPartitions, numReplicationFactor, topicconfig)
This will work because KafkaServerTestHarness (parent of IntegrationTestHarness) has this function which do creation of admin client for you.
(same comment for other tests)
There was a problem hiding this comment.
KafkaServerTestHarness#createTopic uses zkClient to create/update topic configs in ZK mode so we have to wait for the metadata to propagate to all the brokers.
TestUtils.waitForPartitionMetadata(...)
Since, direct ZK client usage is deprecated not using the createTopic method.
| } | ||
|
|
||
| /** | ||
| * Validates the values of the given properties. Should be called only by the broker. |
There was a problem hiding this comment.
thank you for adding these. very useful!
|
accidentally approved this PR earlier, please ignore that. I am still waiting for another revision after the latest round of comments. |
| * TopicConfig#REMOTE_LOG_STORAGE_ENABLE_CONFIG and the default values should be extracted from the KafkaConfig. | ||
| * @param props The properties to be validated | ||
| */ | ||
| public static void validateDefaultValuesInBroker(Map<?, ?> props) { |
There was a problem hiding this comment.
I don't think this is to validate default value in broker, it should be validating broker configs, including user overriding configs. Is that right? Maybe validateConfiguredValuesInBroker?
| .define(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, BOOLEAN, | ||
| RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE, null, MEDIUM, | ||
| RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC); |
There was a problem hiding this comment.
Will this appear in the official doc under Topic Config section? If so, maybe defineInternal?
There was a problem hiding this comment.
Also, we should add a comment to say why we add this broker level config in log Config.
| /** | ||
| * Validates the default values of the LogConfig. Should be called only by the broker. | ||
| * The `props` supplied should contain all the LogConfig properties except | ||
| * TopicConfig#REMOTE_LOG_STORAGE_ENABLE_CONFIG and the default values should be extracted from the KafkaConfig. |
There was a problem hiding this comment.
From what I saw, we will include TopicConfig#REMOTE_LOG_STORAGE_ENABLE_CONFIG in the props parameter. What does this comment mean?
Maybe what you want to say is in the validateDefaultValuesInBroker method, the props doesn't contain any topic-level configs, only broker-level configs, is that right?
There was a problem hiding this comment.
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?
|
Also, could you add PR description to mention what you have done in this PR, and what you didn't do in this PR? Thanks. |
|
Addressed your review comments. PTAL. |
|
Please correct me if I have understood this wrongly, but this pull request proposes propagating a configuration known only to the RemoteLogManagerConfig to the LogConfig via the KafkaConfig? If this is true, I am not certain I approve of this approach. I would argue that we need a new abstraction which compares hierarchically set values. In other parts of the codebase, for example log.segment.bytes and segment.bytes, it is the KafkaConfig which pulls values from the LogConfig rather than push them down. I would prefer that if we are keen on proceeding with adding this validation that it is added in a neutral-zone for now similar to what I did in #14161 (comment). This in my head makes it very clear that we do not yet have the right abstraction for it, but we definitely do not try to nudge already existing abstractions. |
@clolov |
I agree it'd be great if we could find a neutral-zone for such validation, but it doesn't exist currently. It needs some refactor to make that happen. Before that, I think I'm in favor of current approach. I just left some more comments to try not to make the |
| /** | ||
| * Validates the configured values of the LogConfig. Should be called only by the broker. | ||
| * The `props` supplied doesn't contain any topic-level configs, only broker-level configs. | ||
| * The default values should be extracted from the KafkaConfig. | ||
| * @param props The properties to be validated | ||
| */ | ||
| public static void validateConfiguredValuesInBroker(Map<?, ?> props) { |
There was a problem hiding this comment.
Could you explain more about what's the difference between this method and validateValuesInBroker method? I think this method is only invoked when initializing LogManager without topic related configs, is that right? From the method name, it's really difficult to identify what they are doing. Thanks.
| // RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP is defined here to ensure that when system | ||
| // level remote storage functionality is disabled, topics cannot be configured to use remote storage. | ||
| .defineInternal(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, BOOLEAN, | ||
| RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE, null, MEDIUM, | ||
| RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC); |
There was a problem hiding this comment.
This still makes me feel it's not the correct place to put, while other configs are topic related configs. Could we pass in the KafkaConfig object into LogConfig.validate, i.e.
# currently
LogConfig.validate(config, kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()))
# my suggestion
LogConfig.validate(config, kafkaConfig)
so that we don't need to make this LogConfig complicated? Thoughts @kamalcph @divijvaidya ?
There was a problem hiding this comment.
KafkaConfig class is in core module. storage module doesn't have dependency on the core. If we have to do the below change, then we have to make core as dependency for storage. Let me know if it is OK to add it:
public static LogConfig validateAndExtractLogConfig(kafka.server.KafkaConfig kafkaConfig) {
Properties props = kafkaConfig.extractLogConfigMap();
validateValues(props);
if (kafkaConfig.isRemoteLogStorageSystemEnabled()) {
validateRemoteStorageRetentionSize(props);
validateRemoteStorageRetentionTime(props);
}
return new LogConfig(props);
}
There was a problem hiding this comment.
I remembered there will be cycle dependency issue there. So, if we only care about REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP value currently, could we just pass that into LogConfig.validate? I.e. LogConfig.validate(config, sysRemoteStorageEnabled), or like what we did now, but add one more parameter:
LogConfig.validate(config, kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()), sysRemoteStorageEnabled)
There was a problem hiding this comment.
I disagree that we just pass one specific config instead of generic kafkaConfig object, but in the interest of wrapping up this PR since it has undergone multiple revision, I am ok with this approach. We can always come back and refactor later.
|
@kamalcph Tests from this PR are failing in CI such as |
showuon
left a comment
showuon
left a comment
There was a problem hiding this comment.
LGTM! Will merge it after CI build completed. Thanks for the PR!
|
@kamalcph tests are still failing RemoteTopicCrudTest.testCreateRemoteTopicWithValidRetentionTime(String).quorum=kraft Please check. |
|
|
||
| private def verifyRemoteLogTopicConfigs(topicConfig: Properties): Unit = { | ||
| val logBuffer = brokers.flatMap(_.logManager.getLog(new TopicPartition(testTopicName, 0))) | ||
| assertTrue(logBuffer.nonEmpty) |
There was a problem hiding this comment.
The test fails at this place. Do you know how to fix it? Not sure why the log will be empty. The test is being failed for Kraft mode.
at kafka.admin.RemoteTopicCrudTest.verifyRemoteLogTopicConfigs(RemoteTopicCrudTest.scala:287)
There was a problem hiding this comment.
I will take a look tomorrow.
There was a problem hiding this comment.
Please add the following line to TestUtils.createTopicWithAdmin(). This is because we currently only validate that metadata is consistent but after receiving metadata a broker has to process the metadata change as well which is async. Adding the following will ensure that createTopicWithAdmin only returns when log for the topic has indeed been created.
TestUtils.waitUntilTrue(() => logManager.getLog(topicPartition).isDefined, "$testTopicName should be created")
|
Unrelated test failures: |
4 participants
If system level remote storage is not enabled, then enabling remote storage on a topic should throw exception while validating the configs.
More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.
Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.
Committer Checklist (excluded from commit message)