KAFKA-15295: Add config validation when remote storage is enabled on a topic #14176
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,262 @@ | ||
| /** | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package kafka.admin | ||
|
|
||
| import kafka.api.IntegrationTestHarness | ||
| import kafka.server.KafkaConfig | ||
| import kafka.utils.{TestInfoUtils, TestUtils} | ||
| import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry} | ||
| import org.apache.kafka.common.config.{ConfigResource, TopicConfig} | ||
| import org.apache.kafka.common.errors.InvalidConfigurationException | ||
| import org.apache.kafka.server.log.remote.storage.{NoOpRemoteLogMetadataManager, NoOpRemoteStorageManager, RemoteLogManagerConfig} | ||
| import org.junit.jupiter.api.Assertions._ | ||
| import org.junit.jupiter.api.function.Executable | ||
| import org.junit.jupiter.api.{BeforeEach, Tag, TestInfo} | ||
| import org.junit.jupiter.params.ParameterizedTest | ||
| import org.junit.jupiter.params.provider.ValueSource | ||
|
|
||
| import java.util | ||
| import java.util.{Collections, Properties} | ||
| import scala.collection.Seq | ||
| import scala.concurrent.ExecutionException | ||
| import scala.util.Random | ||
|
|
||
| @Tag("integration") | ||
| class RemoteTopicCRUDTest extends IntegrationTestHarness { | ||
|
kamalcph marked this conversation as resolved.
Outdated
kamalcph marked this conversation as resolved.
Outdated
|
||
|
|
||
| val numPartitions = 2 | ||
| val numReplicationFactor = 2 | ||
| var testTopicName: String = _ | ||
|
|
||
| override protected def brokerCount: Int = 2 | ||
|
|
||
| override protected def modifyConfigs(props: Seq[Properties]): Unit = { | ||
| props.foreach(p => p.putAll(overrideProps())) | ||
| } | ||
|
|
||
| override protected def kraftControllerConfigs(): Seq[Properties] = { | ||
| Seq(overrideProps()) | ||
| } | ||
|
|
||
| @BeforeEach | ||
| override def setUp(info: TestInfo): Unit = { | ||
| super.setUp(info) | ||
| testTopicName = s"${info.getTestMethod.get().getName}-${Random.alphanumeric.take(10).mkString}" | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateRemoteTopicWithValidRetentionTime(quorum: String): Unit = { | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, "200") | ||
| topicConfig.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, "100") | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateRemoteTopicWithValidRetentionSize(quorum: String): Unit = { | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.RETENTION_BYTES_CONFIG, "512") | ||
| topicConfig.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, "256") | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateRemoteTopicWithInheritedLocalRetentionTime(quorum: String): Unit = { | ||
| // inherited local retention ms is 1000 | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, "1001") | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateRemoteTopicWithInheritedLocalRetentionSize(quorum: String): Unit = { | ||
| // inherited local retention bytes is 1024 | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.RETENTION_BYTES_CONFIG, "1025") | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateRemoteTopicWithInvalidRetentionTime(quorum: String): Unit = { | ||
| // inherited local retention ms is 1000 | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, "200") | ||
| assertThrowsException(classOf[InvalidConfigurationException], () => | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig)) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateRemoteTopicWithInvalidRetentionSize(quorum: String): Unit = { | ||
| // inherited local retention bytes is 1024 | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.RETENTION_BYTES_CONFIG, "512") | ||
| assertThrowsException(classOf[InvalidConfigurationException], () => | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig)) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testCreateCompactedRemoteStorage(quorum: String): Unit = { | ||
| val topicConfig = new Properties() | ||
| topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") | ||
| topicConfig.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact") | ||
| assertThrowsException(classOf[InvalidConfigurationException], () => | ||
| TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig)) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testEnableRemoteLogOnExistingTopicTest(quorum: String): Unit = { | ||
| val admin = createAdminClient() | ||
| val topicConfig = new Properties() | ||
| TestUtils.createTopicWithAdmin(admin, testTopicName, brokers, numPartitions, numReplicationFactor, | ||
| topicConfig = topicConfig) | ||
|
|
||
| val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]() | ||
| configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName), | ||
| Collections.singleton( | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"), | ||
| AlterConfigOp.OpType.SET)) | ||
| ) | ||
| admin.incrementalAlterConfigs(configs).all().get() | ||
|
Member
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: (same for other tests)
Contributor
Author
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.
Contributor
Author
There was a problem hiding this comment. Addressed with latest commit. |
||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testUpdateTopicConfigWithValidRetentionTimeTest(quorum: String): Unit = { | ||
| 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) | ||
|
|
||
| val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]() | ||
| configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName), | ||
| util.Arrays.asList( | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "200"), | ||
| AlterConfigOp.OpType.SET), | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, "100"), | ||
| AlterConfigOp.OpType.SET) | ||
| )) | ||
| admin.incrementalAlterConfigs(configs).all().get() | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testUpdateTopicConfigWithValidRetentionSizeTest(quorum: String): Unit = { | ||
| 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) | ||
|
|
||
| val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]() | ||
| configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName), | ||
| util.Arrays.asList( | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_BYTES_CONFIG, "200"), | ||
| AlterConfigOp.OpType.SET), | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, "100"), | ||
| AlterConfigOp.OpType.SET) | ||
| )) | ||
| admin.incrementalAlterConfigs(configs).all().get() | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testUpdateTopicConfigWithInheritedLocalRetentionTime(quorum: String): Unit = { | ||
| 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) | ||
|
Comment on lines
+200
to
+
Member
There was a problem hiding this comment. Could be replaced with: This will work because KafkaServerTestHarness (parent of IntegrationTestHarness) has this function which do creation of admin client for you. (same comment for other tests)
Contributor
Author
There was a problem hiding this comment. KafkaServerTestHarness#createTopic uses Since, direct ZK client usage is deprecated not using the |
||
|
|
||
| // inherited local retention ms is 1000 | ||
| val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]() | ||
| configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName), | ||
| util.Arrays.asList( | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "200"), | ||
| AlterConfigOp.OpType.SET), | ||
| )) | ||
| assertThrowsException(classOf[InvalidConfigurationException], | ||
| () => admin.incrementalAlterConfigs(configs).all().get()) | ||
| } | ||
|
|
||
| @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testUpdateTopicConfigWithInheritedLocalRetentionSize(quorum: String): Unit = { | ||
| 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) | ||
|
|
||
| // inherited local retention bytes is 1024 | ||
| val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]() | ||
| configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName), | ||
| util.Arrays.asList( | ||
| new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_BYTES_CONFIG, "512"), | ||
| AlterConfigOp.OpType.SET), | ||
| )) | ||
| assertThrowsException(classOf[InvalidConfigurationException], | ||
| () => admin.incrementalAlterConfigs(configs).all().get(), "Invalid local retention size") | ||
| } | ||
|
|
||
| private def assertThrowsException(exceptionType: Class[_ <: Throwable], | ||
| executable: Executable, | ||
| message: String = ""): Throwable = { | ||
| assertThrows(exceptionType, () => { | ||
| try { | ||
| executable.execute() | ||
| } catch { | ||
| case e: ExecutionException => throw e.getCause | ||
| } | ||
| }, message) | ||
| } | ||
|
|
||
| private def overrideProps(): Properties = { | ||
| val props = new Properties() | ||
| props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true") | ||
| props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, classOf[NoOpRemoteStorageManager].getName) | ||
| props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP, | ||
| classOf[NoOpRemoteLogMetadataManager].getName) | ||
|
|
||
| props.put(KafkaConfig.LogRetentionTimeMillisProp, "2000") | ||
| props.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "1000") | ||
| props.put(KafkaConfig.LogRetentionBytesProp, "2048") | ||
| props.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "1024") | ||
| props | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
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.brokerConfigvssharedServer.controllerConfig? Seems to be the same.Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sharedServerrepresents 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
ControllerConfigurationValidatorbefore 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 atkafka/core/src/main/scala/kafka/server/ConfigAdminManager.scala
Line 155 in f137da0
There was a problem hiding this comment.
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:
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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
There was a problem hiding this comment.
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.