From c56142ff77671504c347180d64ea7339cfe185ac Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Wed, 9 Aug 2023 18:37:08 +0530 Subject: [PATCH 1/9] KAFKA-15295: Add config validation when remote storage is enabled on a topic --- .../src/main/scala/kafka/log/LogManager.scala | 2 +- .../ControllerConfigurationValidator.scala | 4 +- .../scala/kafka/server/ControllerServer.scala | 2 +- .../main/scala/kafka/server/KafkaConfig.scala | 2 + .../scala/kafka/server/ZkAdminManager.scala | 2 +- .../main/scala/kafka/zk/AdminZkClient.scala | 11 +- .../kafka/admin/RemoteTopicCRUDTest.scala | 262 ++++++++++++++++++ .../kafka/server/QuorumTestHarness.scala | 2 +- .../scala/unit/kafka/log/LogConfigTest.scala | 102 ++++++- ...ControllerConfigurationValidatorTest.scala | 12 +- .../unit/kafka/server/KafkaConfigTest.scala | 2 + .../scala/unit/kafka/utils/TestUtils.scala | 4 +- .../config/ServerTopicConfigSynonyms.java | 3 +- .../storage/internals/log/LogConfig.java | 114 +++++--- 14 files changed, 454 insertions(+), 70 deletions(-) create mode 100644 core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 913661fd4f494..9dde6bafc3841 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -1376,7 +1376,7 @@ object LogManager { keepPartitionMetadataFile: Boolean): LogManager = { val defaultProps = config.extractLogConfigMap - LogConfig.validateValues(defaultProps) + LogConfig.validateValuesInBroker(defaultProps) val defaultLogConfig = new LogConfig(defaultProps) val cleanerConfig = LogCleaner.cleanerConfig(config) diff --git a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala index d567596b81f4b..9bc10eba3d39c 100644 --- a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala +++ b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala @@ -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 = { @@ -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) case BROKER => validateBrokerName(resource.name()) case _ => throwExceptionForUnknownResourceType(resource) } diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala index 78045a6985dcb..e9d72ce708434 100644 --- a/core/src/main/scala/kafka/server/ControllerServer.scala +++ b/core/src/main/scala/kafka/server/ControllerServer.scala @@ -231,7 +231,7 @@ class ControllerServer( setMetrics(quorumControllerMetrics). setCreateTopicPolicy(createTopicPolicy.asJava). setAlterConfigPolicy(alterConfigPolicy.asJava). - setConfigurationValidator(new ControllerConfigurationValidator()). + setConfigurationValidator(new ControllerConfigurationValidator(sharedServer.brokerConfig)). setStaticConfig(config.originals). setBootstrapMetadata(bootstrapMetadata). setFatalFaultHandler(sharedServer.fatalQuorumControllerFaultHandler). diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index ddc206f7ed8ca..c9609b1917713 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -2453,6 +2453,8 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami logProps.put(TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_CONFIG, logMessageDownConversionEnable: java.lang.Boolean) logProps.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, logLocalRetentionMs) logProps.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, logLocalRetentionBytes) + logProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, + remoteLogManagerConfig.enableRemoteStorageSystem(): java.lang.Boolean) logProps } diff --git a/core/src/main/scala/kafka/server/ZkAdminManager.scala b/core/src/main/scala/kafka/server/ZkAdminManager.scala index 90fbd7f39962e..ac7a82dc0dfa9 100644 --- a/core/src/main/scala/kafka/server/ZkAdminManager.scala +++ b/core/src/main/scala/kafka/server/ZkAdminManager.scala @@ -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 = diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala index d16394cab05b7..4bea5fe894ba2 100644 --- a/core/src/main/scala/kafka/zk/AdminZkClient.scala +++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala @@ -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} @@ -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 @@ -159,7 +160,7 @@ 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())) } private def writeTopicPartitionAssignment(topic: String, replicaAssignment: Map[Int, ReplicaAssignment], @@ -475,7 +476,7 @@ 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())) } /** diff --git a/core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala b/core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala new file mode 100644 index 0000000000000..88bc18e35912c --- /dev/null +++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala @@ -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 { + + 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() + } + + @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) + + // 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 + } +} diff --git a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala index 0a2cf270bf909..06e2080934d5d 100755 --- a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala +++ b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala @@ -383,7 +383,7 @@ abstract class QuorumTestHarness extends Logging { Time.SYSTEM, name = "ZooKeeperTestHarness", new ZKClientConfig) - adminZkClient = new AdminZkClient(zkClient) + adminZkClient = new AdminZkClient(zkClient, None) } catch { case t: Throwable => CoreUtils.swallow(zookeeper.shutdown(), this) diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index ec64ca82f6992..14eb8a5b31489 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -29,6 +29,8 @@ import java.util.{Collections, Properties} import org.apache.kafka.server.common.MetadataVersion.IBP_3_0_IV1 import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig import org.apache.kafka.storage.internals.log.{LogConfig, ThrottledReplicaListValidator} +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.ValueSource import scala.annotation.nowarn import scala.jdk.CollectionConverters._ @@ -275,27 +277,103 @@ class LogConfigTest { doTestInvalidLocalLogRetentionProps(2000L, -1, 100, 1000L) } - private def doTestInvalidLocalLogRetentionProps(localRetentionMs: Long, localRetentionBytes: Int, retentionBytes: Int, retentionMs: Long) = { + private def doTestInvalidLocalLogRetentionProps(localRetentionMs: Long, + localRetentionBytes: Int, + retentionBytes: Int, + retentionMs: Long) = { + val kafkaProps = TestUtils.createDummyBrokerConfig() + kafkaProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true") + val kafkaConfig = KafkaConfig.fromProps(kafkaProps) + val props = new Properties() + props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") props.put(TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes.toString) props.put(TopicConfig.RETENTION_MS_CONFIG, retentionMs.toString) props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs.toString) props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes.toString) - assertThrows(classOf[ConfigException], () => LogConfig.validate(props)) + assertThrows(classOf[ConfigException], () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap)) } @Test def testEnableRemoteLogStorageOnCompactedTopic(): Unit = { - val props = new Properties() - props.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE) - props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") - LogConfig.validate(props) - props.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT) - assertThrows(classOf[ConfigException], () => LogConfig.validate(props)) - props.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete, compact") - assertThrows(classOf[ConfigException], () => LogConfig.validate(props)) - props.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact, delete") - assertThrows(classOf[ConfigException], () => LogConfig.validate(props)) + val kafkaProps = TestUtils.createDummyBrokerConfig() + kafkaProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true") + val kafkaConfig = KafkaConfig.fromProps(kafkaProps) + + val logProps = new Properties() + logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE) + logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + + logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT) + assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact") + assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete") + assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + } + + @ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}") + @ValueSource(booleans = Array(true, false)) + def testEnableRemoteLogStorage(sysRemoteStorageEnabled: Boolean): Unit = { + val kafkaProps = TestUtils.createDummyBrokerConfig() + kafkaProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) + val kafkaConfig = KafkaConfig.fromProps(kafkaProps) + + val logProps = new Properties() + logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") + if (sysRemoteStorageEnabled) { + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + } else { + val message = assertThrows(classOf[ConfigException], + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + assertTrue(message.getMessage.contains("Tiered Storage functionality is disabled in the broker")) + } + } + + @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionTime with sysRemoteStorageEnabled: {0}") + @ValueSource(booleans = Array(true, false)) + def testTopicCreationWithInvalidRetentionTime(sysRemoteStorageEnabled: Boolean): Unit = { + val kafkaProps = TestUtils.createDummyBrokerConfig() + kafkaProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) + kafkaProps.put(KafkaConfig.LogRetentionTimeMillisProp, "1000") + kafkaProps.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "900") + val kafkaConfig = KafkaConfig.fromProps(kafkaProps) + + // Topic local log retention time inherited from Broker is greater than the topic's complete log retention time + val logProps = new Properties() + logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, sysRemoteStorageEnabled.toString) + logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500") + if (sysRemoteStorageEnabled) { + val message = assertThrows(classOf[ConfigException], + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG)) + } else { + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + } + } + + @ParameterizedTest(name = "testTopicCreationWithInvalidRetentionSize with sysRemoteStorageEnabled: {0}") + @ValueSource(booleans = Array(true, false)) + def testTopicCreationWithInvalidRetentionSize(sysRemoteStorageEnabled: Boolean): Unit = { + val props = TestUtils.createDummyBrokerConfig() + props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) + props.put(KafkaConfig.LogRetentionBytesProp, "1024") + props.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "512") + val kafkaConfig = KafkaConfig.fromProps(props) + + // Topic local retention size inherited from Broker is greater than the topic's complete log retention size + val logProps = new Properties() + logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, sysRemoteStorageEnabled.toString) + logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128") + if (sysRemoteStorageEnabled) { + val message = assertThrows(classOf[ConfigException], + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + print(message) + assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) + } else { + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + } } } diff --git a/core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala b/core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala index 36a8d71fb9759..2edb663a88e4e 100644 --- a/core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala @@ -17,18 +17,20 @@ package kafka.server -import java.util.TreeMap -import java.util.Collections.emptyMap - -import org.junit.jupiter.api.Test +import kafka.utils.TestUtils import org.apache.kafka.common.config.ConfigResource import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER, TOPIC} import org.apache.kafka.common.config.TopicConfig.{SEGMENT_BYTES_CONFIG, SEGMENT_JITTER_MS_CONFIG, SEGMENT_MS_CONFIG} import org.apache.kafka.common.errors.{InvalidConfigurationException, InvalidRequestException, InvalidTopicException} import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows} +import org.junit.jupiter.api.Test + +import java.util.Collections.emptyMap +import java.util.TreeMap class ControllerConfigurationValidatorTest { - val validator = new ControllerConfigurationValidator() + val config = new KafkaConfig(TestUtils.createDummyBrokerConfig()) + val validator = new ControllerConfigurationValidator(config) @Test def testDefaultTopicResourceIsRejected(): Unit = { diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 5136862192cf7..4f8323fb0c6cb 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -1124,6 +1124,8 @@ class KafkaConfigTest { // topic only config case LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG => // topic only config + case RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP => + // not dynamically updatable case prop => fail(prop + " must be explicitly checked for dynamic updatability. Note that LogConfig(s) require that KafkaConfig value lookups are dynamic and not static values.") } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 08cfffdb7842a..5f391ea17e92b 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -561,7 +561,7 @@ object TestUtils extends Logging { replicationFactor: Int = 1, servers: Seq[KafkaBroker], topicConfig: Properties = new Properties): scala.collection.immutable.Map[Int, Int] = { - val adminZkClient = new AdminZkClient(zkClient) + val adminZkClient = new AdminZkClient(zkClient, None) // create topic waitUntilTrue( () => { var hasSessionExpirationException = false @@ -605,7 +605,7 @@ object TestUtils extends Logging { partitionReplicaAssignment: collection.Map[Int, Seq[Int]], servers: Seq[KafkaBroker], topicConfig: Properties): scala.collection.immutable.Map[Int, Int] = { - val adminZkClient = new AdminZkClient(zkClient) + val adminZkClient = new AdminZkClient(zkClient, None) // create topic waitUntilTrue( () => { var hasSessionExpirationException = false diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java index ece3891a419e4..f6041f01f03de 100644 --- a/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java +++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java @@ -85,7 +85,8 @@ public final class ServerTopicConfigSynonyms { sameNameWithLogPrefix(TopicConfig.MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_CONFIG), sameNameWithLogPrefix(TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_CONFIG), sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG), - sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG) + sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG), + sameName("remote.log.storage.system.enable") )); /** diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index 6ed1e8483735b..f8e2302163e01 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -57,6 +57,7 @@ import org.apache.kafka.server.common.MetadataVersion; import org.apache.kafka.server.common.MetadataVersionValidator; import org.apache.kafka.server.config.ServerTopicConfigSynonyms; +import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig; import org.apache.kafka.server.record.BrokerCompressionType; public class LogConfig extends AbstractConfig { @@ -265,7 +266,10 @@ public Optional serverConfigName(String configName) { .define(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, LONG, DEFAULT_LOCAL_RETENTION_MS, atLeast(-2), MEDIUM, TopicConfig.LOCAL_LOG_RETENTION_MS_DOC) .define(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, LONG, DEFAULT_LOCAL_RETENTION_BYTES, atLeast(-2), MEDIUM, - TopicConfig.LOCAL_LOG_RETENTION_BYTES_DOC); + TopicConfig.LOCAL_LOG_RETENTION_BYTES_DOC) + .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); } public final Set overriddenConfigs; @@ -459,49 +463,69 @@ public static void validateValues(Map props) { long maxCompactionLag = (Long) props.get(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG); if (minCompactionLag > maxCompactionLag) { throw new InvalidConfigurationException("conflict topic config setting " - + TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG + " (" + minCompactionLag + ") > " - + TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG + " (" + maxCompactionLag + ")"); + + TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG + " (" + minCompactionLag + ") > " + + TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG + " (" + maxCompactionLag + ")"); } + } - if (props.containsKey(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG)) { - boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); - String cleanupPolicy = props.get(TopicConfig.CLEANUP_POLICY_CONFIG).toString().toLowerCase(Locale.getDefault()); - if (isRemoteStorageEnabled && cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_COMPACT)) { - throw new ConfigException("Remote log storage is unsupported for the compacted topics"); - } + public static void validateValuesInBroker(Map props) { + validateValues(props); + validateRemoteStorageOnlyIfSystemEnabled(props); + validateNoRemoteStorageForCompactedTopic(props); + validateRemoteStorageRetentionSize(props); + validateRemoteStorageRetentionTime(props); + } + + private static void validateRemoteStorageOnlyIfSystemEnabled(Map props) { + Boolean isRemoteLogStorageSystemEnabled = + (Boolean) props.get(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP); + Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); + if (!isRemoteLogStorageSystemEnabled && isRemoteStorageEnabled) { + throw new ConfigException("Tiered Storage functionality is disabled in the broker. " + + "Topic cannot be configured with remote log storage."); + } + } + + private static void validateNoRemoteStorageForCompactedTopic(Map props) { + String cleanupPolicy = props.get(TopicConfig.CLEANUP_POLICY_CONFIG).toString().toLowerCase(Locale.getDefault()); + Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); + if (isRemoteStorageEnabled && cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_COMPACT)) { + throw new ConfigException("Remote log storage is unsupported for the compacted topics"); } + } - if (props.containsKey(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) { - Long retentionBytes = (Long) props.get(TopicConfig.RETENTION_BYTES_CONFIG); - Long localLogRetentionBytes = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG); - if (retentionBytes > -1 && localLogRetentionBytes != -2) { - if (localLogRetentionBytes == -1) { - String message = String.format("Value must not be -1 as %s value is set as %d.", - TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes); - throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localLogRetentionBytes, message); - } - if (localLogRetentionBytes > retentionBytes) { - String message = String.format("Value must not be more than %s property value: %d", - TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes); - throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localLogRetentionBytes, message); - } + private static void validateRemoteStorageRetentionSize(Map props) { + Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); + Long retentionBytes = (Long) props.get(TopicConfig.RETENTION_BYTES_CONFIG); + Long localRetentionBytes = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG); + if (isRemoteStorageEnabled && retentionBytes > -1 && localRetentionBytes != -2) { + if (localRetentionBytes == -1) { + String message = String.format("Value must not be -1 as %s value is set as %d.", + TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes); + throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes, message); + } + if (localRetentionBytes > retentionBytes) { + String message = String.format("Value must not be more than %s property value: %d", + TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes); + throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes, message); } } + } - if (props.containsKey(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG)) { - Long retentionMs = (Long) props.get(TopicConfig.RETENTION_MS_CONFIG); - Long localLogRetentionMs = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG); - if (retentionMs != -1 && localLogRetentionMs != -2) { - if (localLogRetentionMs == -1) { - String message = String.format("Value must not be -1 as %s value is set as %d.", - TopicConfig.RETENTION_MS_CONFIG, retentionMs); - throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localLogRetentionMs, message); - } - if (localLogRetentionMs > retentionMs) { - String message = String.format("Value must not be more than %s property value: %d", - TopicConfig.RETENTION_MS_CONFIG, retentionMs); - throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localLogRetentionMs, message); - } + private static void validateRemoteStorageRetentionTime(Map props) { + Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); + Long retentionMs = (Long) props.get(TopicConfig.RETENTION_MS_CONFIG); + Long localRetentionMs = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG); + if (isRemoteStorageEnabled && retentionMs != -1 && localRetentionMs != -2) { + if (localRetentionMs == -1) { + String message = String.format("Value must not be -1 as %s value is set as %d.", + TopicConfig.RETENTION_MS_CONFIG, retentionMs); + throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs, message); + } + if (localRetentionMs > retentionMs) { + String message = String.format("Value must not be more than %s property value: %d", + TopicConfig.RETENTION_MS_CONFIG, retentionMs); + throw new ConfigException(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs, message); } } } @@ -510,9 +534,21 @@ public static void validateValues(Map props) { * Check that the given properties contain only valid log config names and that all values can be parsed and are valid */ public static void validate(Properties props) { + validate(props, Collections.emptyMap()); + } + + public static void validate(Properties props, + Map defaultConfigs) { validateNames(props); - Map valueMaps = CONFIG.parse(props); - validateValues(valueMaps); + if (defaultConfigs == null || defaultConfigs.isEmpty()) { + Map valueMaps = CONFIG.parse(props); + validateValues(valueMaps); + } else { + Map configWithBrokerDefaults = new HashMap<>(defaultConfigs); + configWithBrokerDefaults.putAll(props); + Map valueMaps = CONFIG.parse(configWithBrokerDefaults); + validateValuesInBroker(valueMaps); + } } public static void main(String[] args) { From e184917801b86d1447ff2c1ab2af487939b7aefb Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Sun, 13 Aug 2023 13:29:21 +0530 Subject: [PATCH 2/9] Fix the test failures. --- .../src/main/scala/kafka/log/LogManager.scala | 2 +- .../scala/unit/kafka/log/LogConfigTest.scala | 22 ++++++++- .../storage/internals/log/LogConfig.java | 48 ++++++++++++++----- 3 files changed, 58 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 9dde6bafc3841..f40334c7efbe6 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -1376,7 +1376,7 @@ object LogManager { keepPartitionMetadataFile: Boolean): LogManager = { val defaultProps = config.extractLogConfigMap - LogConfig.validateValuesInBroker(defaultProps) + LogConfig.validateDefaultValuesInBroker(defaultProps) val defaultLogConfig = new LogConfig(defaultProps) val cleanerConfig = LogCleaner.cleanerConfig(config) diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index 14eb8a5b31489..68d949725225f 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -105,6 +105,9 @@ class LogConfigTest { props.setProperty(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100") props.setProperty(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200") assertThrows(classOf[Exception], () => LogConfig.validate(props)) + assertThrows(classOf[Exception], () => LogConfig.validateValues(props)) + assertThrows(classOf[Exception], () => LogConfig.validateDefaultValuesInBroker(props)) + assertThrows(classOf[Exception], () => LogConfig.validateValuesInBroker(props)) } @Test @@ -370,10 +373,27 @@ class LogConfigTest { if (sysRemoteStorageEnabled) { val message = assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) - print(message) assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) } else { LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) } } + + @ParameterizedTest(name = "testValidateDefaultLogConfigs with sysRemoteStorageEnabled: {0}") + @ValueSource(booleans = Array(true, false)) + def testValidateDefaultLogConfigs(sysRemoteStorageEnabled: Boolean): Unit = { + val props = TestUtils.createDummyBrokerConfig() + props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) + props.put(KafkaConfig.LogRetentionBytesProp, "1024") + props.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "2048") + val kafkaConfig = KafkaConfig.fromProps(props) + + if (sysRemoteStorageEnabled) { + val message = assertThrows(classOf[ConfigException], + () => LogConfig.validateDefaultValuesInBroker(kafkaConfig.extractLogConfigMap)) + assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) + } else { + LogConfig.validateDefaultValuesInBroker(kafkaConfig.extractLogConfigMap) + } + } } diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index f8e2302163e01..5f950b14e1672 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -458,6 +458,11 @@ public static void validateNames(Properties props) { throw new InvalidConfigurationException("Unknown topic config name: " + name); } + /** + * Validates the values of the given properties. Can be called by both client and server. + * The props passed is a subset of the topic configs and won't contain all the LogConfig properties. + * @param props The properties to be validated + */ public static void validateValues(Map props) { long minCompactionLag = (Long) props.get(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG); long maxCompactionLag = (Long) props.get(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG); @@ -468,19 +473,41 @@ public static void validateValues(Map props) { } } + /** + * Validates the default values of the given properties. Should be called only by the broker. + * The props passed should contain all the LogConfig properties except TopicConfig#REMOTE_LOG_STORAGE_ENABLE_CONFIG + * @param props The properties to be validated + */ + public static void validateDefaultValuesInBroker(Map props) { + validateValues(props); + Boolean isRemoteLogStorageSystemEnabled = + (Boolean) props.get(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP); + if (isRemoteLogStorageSystemEnabled) { + validateRemoteStorageRetentionSize(props); + validateRemoteStorageRetentionTime(props); + } + } + + /** + * Validates the values of the given properties. Should be called only by the broker. + * The props passed should contain all the LogConfig properties. + * @param props The properties to be validated + */ public static void validateValuesInBroker(Map props) { validateValues(props); - validateRemoteStorageOnlyIfSystemEnabled(props); - validateNoRemoteStorageForCompactedTopic(props); - validateRemoteStorageRetentionSize(props); - validateRemoteStorageRetentionTime(props); + Boolean isRemoteLogStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); + if (isRemoteLogStorageEnabled) { + validateRemoteStorageOnlyIfSystemEnabled(props); + validateNoRemoteStorageForCompactedTopic(props); + validateRemoteStorageRetentionSize(props); + validateRemoteStorageRetentionTime(props); + } } private static void validateRemoteStorageOnlyIfSystemEnabled(Map props) { Boolean isRemoteLogStorageSystemEnabled = (Boolean) props.get(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP); - Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); - if (!isRemoteLogStorageSystemEnabled && isRemoteStorageEnabled) { + if (!isRemoteLogStorageSystemEnabled) { throw new ConfigException("Tiered Storage functionality is disabled in the broker. " + "Topic cannot be configured with remote log storage."); } @@ -488,17 +515,15 @@ private static void validateRemoteStorageOnlyIfSystemEnabled(Map props) { private static void validateNoRemoteStorageForCompactedTopic(Map props) { String cleanupPolicy = props.get(TopicConfig.CLEANUP_POLICY_CONFIG).toString().toLowerCase(Locale.getDefault()); - Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); - if (isRemoteStorageEnabled && cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_COMPACT)) { + if (cleanupPolicy.contains(TopicConfig.CLEANUP_POLICY_COMPACT)) { throw new ConfigException("Remote log storage is unsupported for the compacted topics"); } } private static void validateRemoteStorageRetentionSize(Map props) { - Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); Long retentionBytes = (Long) props.get(TopicConfig.RETENTION_BYTES_CONFIG); Long localRetentionBytes = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG); - if (isRemoteStorageEnabled && retentionBytes > -1 && localRetentionBytes != -2) { + if (retentionBytes > -1 && localRetentionBytes != -2) { if (localRetentionBytes == -1) { String message = String.format("Value must not be -1 as %s value is set as %d.", TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes); @@ -513,10 +538,9 @@ private static void validateRemoteStorageRetentionSize(Map props) { } private static void validateRemoteStorageRetentionTime(Map props) { - Boolean isRemoteStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); Long retentionMs = (Long) props.get(TopicConfig.RETENTION_MS_CONFIG); Long localRetentionMs = (Long) props.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG); - if (isRemoteStorageEnabled && retentionMs != -1 && localRetentionMs != -2) { + if (retentionMs != -1 && localRetentionMs != -2) { if (localRetentionMs == -1) { String message = String.format("Value must not be -1 as %s value is set as %d.", TopicConfig.RETENTION_MS_CONFIG, retentionMs); From 1ecca4ecd286ee87e4c560f7d0ec5e486bd8357c Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Sun, 13 Aug 2023 13:53:32 +0530 Subject: [PATCH 3/9] update the comments for clarity --- .../apache/kafka/storage/internals/log/LogConfig.java | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index 5f950b14e1672..36a6bdecf3ace 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -460,7 +460,8 @@ public static void validateNames(Properties props) { /** * Validates the values of the given properties. Can be called by both client and server. - * The props passed is a subset of the topic configs and won't contain all the LogConfig properties. + * The `props` supplied should contain all the LogConfig properties and the default values are extracted from the + * LogConfig class. * @param props The properties to be validated */ public static void validateValues(Map props) { @@ -474,8 +475,9 @@ public static void validateValues(Map props) { } /** - * Validates the default values of the given properties. Should be called only by the broker. - * The props passed should contain all the LogConfig properties except TopicConfig#REMOTE_LOG_STORAGE_ENABLE_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. * @param props The properties to be validated */ public static void validateDefaultValuesInBroker(Map props) { @@ -490,7 +492,8 @@ public static void validateDefaultValuesInBroker(Map props) { /** * Validates the values of the given properties. Should be called only by the broker. - * The props passed should contain all the LogConfig properties. + * The `props` supplied should contain all the LogConfig properties and the default values should be extracted from + * the KafkaConfig. * @param props The properties to be validated */ public static void validateValuesInBroker(Map props) { From 4f15a3a3d2cc694ea9fa5e840e50f22f3c2b2e98 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Mon, 14 Aug 2023 17:13:11 +0530 Subject: [PATCH 4/9] Addressed the review comments. --- .../src/main/scala/kafka/log/LogManager.scala | 2 +- ...UDTest.scala => RemoteTopicCrudTest.scala} | 32 +++++++++++++++++-- .../kafka/server/QuorumTestHarness.scala | 2 +- .../scala/unit/kafka/log/LogConfigTest.scala | 6 ++-- .../scala/unit/kafka/utils/TestUtils.scala | 2 +- .../storage/internals/log/LogConfig.java | 12 ++++--- 6 files changed, 43 insertions(+), 13 deletions(-) rename core/src/test/scala/integration/kafka/admin/{RemoteTopicCRUDTest.scala => RemoteTopicCrudTest.scala} (87%) diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index f40334c7efbe6..8fa764996945d 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -1376,7 +1376,7 @@ object LogManager { keepPartitionMetadataFile: Boolean): LogManager = { val defaultProps = config.extractLogConfigMap - LogConfig.validateDefaultValuesInBroker(defaultProps) + LogConfig.validateConfiguredValuesInBroker(defaultProps) val defaultLogConfig = new LogConfig(defaultProps) val cleanerConfig = LogCleaner.cleanerConfig(config) diff --git a/core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala similarity index 87% rename from core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala rename to core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala index 88bc18e35912c..7cbbac4dc219b 100644 --- a/core/src/test/scala/integration/kafka/admin/RemoteTopicCRUDTest.scala +++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala @@ -36,11 +36,12 @@ import scala.concurrent.ExecutionException import scala.util.Random @Tag("integration") -class RemoteTopicCRUDTest extends IntegrationTestHarness { +class RemoteTopicCrudTest extends IntegrationTestHarness { val numPartitions = 2 val numReplicationFactor = 2 var testTopicName: String = _ + var sysRemoteStorageEnabled = true override protected def brokerCount: Int = 2 @@ -54,6 +55,9 @@ class RemoteTopicCRUDTest extends IntegrationTestHarness { @BeforeEach override def setUp(info: TestInfo): Unit = { + if (info.getTestMethod.get().getName.endsWith("SystemRemoteStorageIsDisabled")) { + sysRemoteStorageEnabled = false + } super.setUp(info) testTopicName = s"${info.getTestMethod.get().getName}-${Random.alphanumeric.take(10).mkString}" } @@ -154,6 +158,30 @@ class RemoteTopicCRUDTest extends IntegrationTestHarness { admin.incrementalAlterConfigs(configs).all().get() } + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk", "kraft")) + def testEnableRemoteLogWhenSystemRemoteStorageIsDisabled(quorum: String): Unit = { + val admin = createAdminClient() + + val topicConfigWithRemoteStorage = new Properties() + topicConfigWithRemoteStorage.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") + val message = assertThrowsException(classOf[InvalidConfigurationException], + () => TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, + numReplicationFactor, topicConfig = topicConfigWithRemoteStorage)) + assertTrue(message.getMessage.contains("Tiered Storage functionality is disabled in the broker")) + + TestUtils.createTopicWithAdmin(admin, testTopicName, brokers, numPartitions, numReplicationFactor) + 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)) + ) + val errorMessage = assertThrowsException(classOf[InvalidConfigurationException], + () => admin.incrementalAlterConfigs(configs).all().get()) + assertTrue(errorMessage.getMessage.contains("Tiered Storage functionality is disabled in the broker")) + } + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @ValueSource(strings = Array("zk", "kraft")) def testUpdateTopicConfigWithValidRetentionTimeTest(quorum: String): Unit = { @@ -248,7 +276,7 @@ class RemoteTopicCRUDTest extends IntegrationTestHarness { private def overrideProps(): Properties = { val props = new Properties() - props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true") + props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) 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) diff --git a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala index 06e2080934d5d..0a2cf270bf909 100755 --- a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala +++ b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala @@ -383,7 +383,7 @@ abstract class QuorumTestHarness extends Logging { Time.SYSTEM, name = "ZooKeeperTestHarness", new ZKClientConfig) - adminZkClient = new AdminZkClient(zkClient, None) + adminZkClient = new AdminZkClient(zkClient) } catch { case t: Throwable => CoreUtils.swallow(zookeeper.shutdown(), this) diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index 68d949725225f..8dd1db34164df 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -106,7 +106,7 @@ class LogConfigTest { props.setProperty(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200") assertThrows(classOf[Exception], () => LogConfig.validate(props)) assertThrows(classOf[Exception], () => LogConfig.validateValues(props)) - assertThrows(classOf[Exception], () => LogConfig.validateDefaultValuesInBroker(props)) + assertThrows(classOf[Exception], () => LogConfig.validateConfiguredValuesInBroker(props)) assertThrows(classOf[Exception], () => LogConfig.validateValuesInBroker(props)) } @@ -390,10 +390,10 @@ class LogConfigTest { if (sysRemoteStorageEnabled) { val message = assertThrows(classOf[ConfigException], - () => LogConfig.validateDefaultValuesInBroker(kafkaConfig.extractLogConfigMap)) + () => LogConfig.validateConfiguredValuesInBroker(kafkaConfig.extractLogConfigMap)) assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) } else { - LogConfig.validateDefaultValuesInBroker(kafkaConfig.extractLogConfigMap) + LogConfig.validateConfiguredValuesInBroker(kafkaConfig.extractLogConfigMap) } } } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 5f391ea17e92b..28db39c84898e 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -561,7 +561,7 @@ object TestUtils extends Logging { replicationFactor: Int = 1, servers: Seq[KafkaBroker], topicConfig: Properties = new Properties): scala.collection.immutable.Map[Int, Int] = { - val adminZkClient = new AdminZkClient(zkClient, None) + val adminZkClient = new AdminZkClient(zkClient) // create topic waitUntilTrue( () => { var hasSessionExpirationException = false diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index 36a6bdecf3ace..a43b577bc1e1f 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -267,7 +267,9 @@ public Optional serverConfigName(String configName) { TopicConfig.LOCAL_LOG_RETENTION_MS_DOC) .define(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, LONG, DEFAULT_LOCAL_RETENTION_BYTES, atLeast(-2), MEDIUM, TopicConfig.LOCAL_LOG_RETENTION_BYTES_DOC) - .define(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, BOOLEAN, + // 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); } @@ -475,12 +477,12 @@ public static void validateValues(Map props) { } /** - * 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. + * 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 validateDefaultValuesInBroker(Map props) { + public static void validateConfiguredValuesInBroker(Map props) { validateValues(props); Boolean isRemoteLogStorageSystemEnabled = (Boolean) props.get(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP); From d979f01d34d5817afbdd732198aab2629bd78c0e Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Mon, 14 Aug 2023 18:11:03 +0530 Subject: [PATCH 5/9] Addressed the review-2 comments. --- .../kafka/admin/RemoteTopicCrudTest.scala | 46 ++++++++++++++++++- .../storage/internals/log/LogConfig.java | 34 ++++++++++++++ 2 files changed, 78 insertions(+), 2 deletions(-) diff --git a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala index 7cbbac4dc219b..218f9a61af8ea 100644 --- a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala +++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala @@ -20,9 +20,11 @@ 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.TopicPartition 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.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} @@ -71,6 +73,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { topicConfig.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, "100") TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, topicConfig = topicConfig) + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -82,6 +85,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { topicConfig.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, "256") TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, topicConfig = topicConfig) + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -93,6 +97,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, "1001") TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, topicConfig = topicConfig) + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -104,6 +109,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { topicConfig.put(TopicConfig.RETENTION_BYTES_CONFIG, "1025") TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, numReplicationFactor, topicConfig = topicConfig) + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -156,6 +162,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { AlterConfigOp.OpType.SET)) ) admin.incrementalAlterConfigs(configs).all().get() + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -200,6 +207,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { AlterConfigOp.OpType.SET) )) admin.incrementalAlterConfigs(configs).all().get() + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -220,6 +228,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { AlterConfigOp.OpType.SET) )) admin.incrementalAlterConfigs(configs).all().get() + verifyRemoteLogTopicConfigs(topicConfig) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -274,10 +283,43 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { }, message) } + private def verifyRemoteLogTopicConfigs(topicConfig: Properties): Unit = { + val logOpt = brokers.head.logManager.getLog(new TopicPartition(testTopicName, 0)) + assertTrue(logOpt.isDefined) + val log = logOpt.get + TestUtils.waitUntilTrue(() => { + var result = true + if (topicConfig.containsKey(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG).toBoolean == + log.config.remoteStorageEnable() + } + if (topicConfig.containsKey(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG).toLong == + log.config.localRetentionBytes() + } + if (topicConfig.containsKey(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG).toLong == log.config.localRetentionMs() + } + if (topicConfig.containsKey(TopicConfig.RETENTION_MS_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.RETENTION_MS_CONFIG).toLong == log.config.retentionMs + } + if (topicConfig.containsKey(TopicConfig.RETENTION_BYTES_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.RETENTION_BYTES_CONFIG).toLong == log.config.retentionSize + } + result + }, s"Failed to update topic config $topicConfig, actual: ${log.config}") + } + private def overrideProps(): Properties = { val props = new Properties() props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) - props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, classOf[NoOpRemoteStorageManager].getName) + 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) diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index a43b577bc1e1f..66622ca3d5499 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -580,6 +580,40 @@ public static void validate(Properties props, } } + @Override + public String toString() { + return "LogConfig{" + + "segmentSize=" + segmentSize + + ", segmentMs=" + segmentMs + + ", segmentJitterMs=" + segmentJitterMs + + ", maxIndexSize=" + maxIndexSize + + ", flushInterval=" + flushInterval + + ", flushMs=" + flushMs + + ", retentionSize=" + retentionSize + + ", retentionMs=" + retentionMs + + ", indexInterval=" + indexInterval + + ", fileDeleteDelayMs=" + fileDeleteDelayMs + + ", deleteRetentionMs=" + deleteRetentionMs + + ", compactionLagMs=" + compactionLagMs + + ", maxCompactionLagMs=" + maxCompactionLagMs + + ", minCleanableRatio=" + minCleanableRatio + + ", compact=" + compact + + ", delete=" + delete + + ", uncleanLeaderElectionEnable=" + uncleanLeaderElectionEnable + + ", minInSyncReplicas=" + minInSyncReplicas + + ", compressionType='" + compressionType + '\'' + + ", preallocate=" + preallocate + + ", messageFormatVersion=" + messageFormatVersion + + ", messageTimestampType=" + messageTimestampType + + ", messageTimestampDifferenceMaxMs=" + messageTimestampDifferenceMaxMs + + ", leaderReplicationThrottledReplicas=" + leaderReplicationThrottledReplicas + + ", followerReplicationThrottledReplicas=" + followerReplicationThrottledReplicas + + ", messageDownConversionEnable=" + messageDownConversionEnable + + ", remoteLogConfig=" + remoteLogConfig + + ", maxMessageSize=" + maxMessageSize + + '}'; + } + public static void main(String[] args) { System.out.println(CONFIG.toHtml(4, config -> "topicconfigs_" + config)); } From 21bf7a3155f1c28f6d8fb40a4c9abecbb2b87b87 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Tue, 15 Aug 2023 11:21:17 +0530 Subject: [PATCH 6/9] Addressed the review comments. --- .../src/main/scala/kafka/log/LogManager.scala | 2 +- .../ControllerConfigurationValidator.scala | 2 +- .../main/scala/kafka/server/KafkaConfig.scala | 2 - .../main/scala/kafka/zk/AdminZkClient.scala | 8 ++- .../scala/unit/kafka/log/LogConfigTest.scala | 37 ++++++------- .../unit/kafka/server/KafkaConfigTest.scala | 2 - .../scala/unit/kafka/utils/TestUtils.scala | 2 +- .../config/ServerTopicConfigSynonyms.java | 3 +- .../storage/internals/log/LogConfig.java | 54 ++++++++++--------- 9 files changed, 57 insertions(+), 55 deletions(-) diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 8fa764996945d..bed9f0dfa03b0 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -1376,7 +1376,7 @@ object LogManager { keepPartitionMetadataFile: Boolean): LogManager = { val defaultProps = config.extractLogConfigMap - LogConfig.validateConfiguredValuesInBroker(defaultProps) + LogConfig.validateBrokerLogConfigValues(defaultProps, config.isRemoteLogStorageSystemEnabled) val defaultLogConfig = new LogConfig(defaultProps) val cleanerConfig = LogCleaner.cleanerConfig(config) diff --git a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala index 9bc10eba3d39c..ccdd0ac31afa4 100644 --- a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala +++ b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala @@ -106,7 +106,7 @@ class ControllerConfigurationValidator(kafkaConfig: KafkaConfig) extends Configu throw new InvalidConfigurationException("Null value not supported for topic configs: " + nullTopicConfigs.mkString(",")) } - LogConfig.validate(properties, kafkaConfig.extractLogConfigMap) + LogConfig.validate(properties, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled) case BROKER => validateBrokerName(resource.name()) case _ => throwExceptionForUnknownResourceType(resource) } diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index c9609b1917713..ddc206f7ed8ca 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -2453,8 +2453,6 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami logProps.put(TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_CONFIG, logMessageDownConversionEnable: java.lang.Boolean) logProps.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, logLocalRetentionMs) logProps.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, logLocalRetentionBytes) - logProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, - remoteLogManagerConfig.enableRemoteStorageSystem(): java.lang.Boolean) logProps } diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala index 4bea5fe894ba2..3a4715660fbc2 100644 --- a/core/src/main/scala/kafka/zk/AdminZkClient.scala +++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala @@ -160,7 +160,9 @@ class AdminZkClient(zkClient: KafkaZkClient, partitionReplicaAssignment.keys.filter(_ >= 0).sum != sequenceSum) throw new InvalidReplicaAssignmentException("partitions should be a consecutive 0-based integer sequence") - LogConfig.validate(config, kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap())) + LogConfig.validate(config, + kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()), + kafkaConfig.exists(_.isRemoteLogStorageSystemEnabled)) } private def writeTopicPartitionAssignment(topic: String, replicaAssignment: Map[Int, ReplicaAssignment], @@ -476,7 +478,9 @@ class AdminZkClient(zkClient: KafkaZkClient, if (!zkClient.topicExists(topic)) throw new UnknownTopicOrPartitionException(s"Topic '$topic' does not exist.") // remove the topic overrides - LogConfig.validate(configs, kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap())) + LogConfig.validate(configs, + kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()), + kafkaConfig.exists(_.isRemoteLogStorageSystemEnabled)) } /** diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index 8dd1db34164df..dd9b20289c4c6 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -105,9 +105,6 @@ class LogConfigTest { props.setProperty(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, "100") props.setProperty(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, "200") assertThrows(classOf[Exception], () => LogConfig.validate(props)) - assertThrows(classOf[Exception], () => LogConfig.validateValues(props)) - assertThrows(classOf[Exception], () => LogConfig.validateConfiguredValuesInBroker(props)) - assertThrows(classOf[Exception], () => LogConfig.validateValuesInBroker(props)) } @Test @@ -295,7 +292,8 @@ class LogConfigTest { props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs.toString) props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes.toString) - assertThrows(classOf[ConfigException], () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap)) + assertThrows(classOf[ConfigException], + () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) } @Test @@ -307,14 +305,17 @@ class LogConfigTest { val logProps = new Properties() logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE) logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") - LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled) logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT) - assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + assertThrows(classOf[ConfigException], + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact") - assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + assertThrows(classOf[ConfigException], + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete") - assertThrows(classOf[ConfigException], () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + assertThrows(classOf[ConfigException], + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) } @ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}") @@ -327,10 +328,10 @@ class LogConfigTest { val logProps = new Properties() logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") if (sysRemoteStorageEnabled) { - LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled) } else { val message = assertThrows(classOf[ConfigException], - () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) assertTrue(message.getMessage.contains("Tiered Storage functionality is disabled in the broker")) } } @@ -350,10 +351,10 @@ class LogConfigTest { logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500") if (sysRemoteStorageEnabled) { val message = assertThrows(classOf[ConfigException], - () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG)) } else { - LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled) } } @@ -372,16 +373,16 @@ class LogConfigTest { logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128") if (sysRemoteStorageEnabled) { val message = assertThrows(classOf[ConfigException], - () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap)) + () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) } else { - LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap) + LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled) } } - @ParameterizedTest(name = "testValidateDefaultLogConfigs with sysRemoteStorageEnabled: {0}") + @ParameterizedTest(name = "testValidateBrokerLogConfigs with sysRemoteStorageEnabled: {0}") @ValueSource(booleans = Array(true, false)) - def testValidateDefaultLogConfigs(sysRemoteStorageEnabled: Boolean): Unit = { + def testValidateBrokerLogConfigs(sysRemoteStorageEnabled: Boolean): Unit = { val props = TestUtils.createDummyBrokerConfig() props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, sysRemoteStorageEnabled.toString) props.put(KafkaConfig.LogRetentionBytesProp, "1024") @@ -390,10 +391,10 @@ class LogConfigTest { if (sysRemoteStorageEnabled) { val message = assertThrows(classOf[ConfigException], - () => LogConfig.validateConfiguredValuesInBroker(kafkaConfig.extractLogConfigMap)) + () => LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)) assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) } else { - LogConfig.validateConfiguredValuesInBroker(kafkaConfig.extractLogConfigMap) + LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled) } } } diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 4f8323fb0c6cb..5136862192cf7 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -1124,8 +1124,6 @@ class KafkaConfigTest { // topic only config case LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG => // topic only config - case RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP => - // not dynamically updatable case prop => fail(prop + " must be explicitly checked for dynamic updatability. Note that LogConfig(s) require that KafkaConfig value lookups are dynamic and not static values.") } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 28db39c84898e..08cfffdb7842a 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -605,7 +605,7 @@ object TestUtils extends Logging { partitionReplicaAssignment: collection.Map[Int, Seq[Int]], servers: Seq[KafkaBroker], topicConfig: Properties): scala.collection.immutable.Map[Int, Int] = { - val adminZkClient = new AdminZkClient(zkClient, None) + val adminZkClient = new AdminZkClient(zkClient) // create topic waitUntilTrue( () => { var hasSessionExpirationException = false diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java index f6041f01f03de..ece3891a419e4 100644 --- a/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java +++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java @@ -85,8 +85,7 @@ public final class ServerTopicConfigSynonyms { sameNameWithLogPrefix(TopicConfig.MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_CONFIG), sameNameWithLogPrefix(TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_CONFIG), sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG), - sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG), - sameName("remote.log.storage.system.enable") + sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG) )); /** diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index 66622ca3d5499..609b0de961fe1 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -57,7 +57,6 @@ import org.apache.kafka.server.common.MetadataVersion; import org.apache.kafka.server.common.MetadataVersionValidator; import org.apache.kafka.server.config.ServerTopicConfigSynonyms; -import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig; import org.apache.kafka.server.record.BrokerCompressionType; public class LogConfig extends AbstractConfig { @@ -112,6 +111,15 @@ private RemoteLogConfig(LogConfig config) { this.localRetentionMs = config.getLong(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG); this.localRetentionBytes = config.getLong(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG); } + + @Override + public String toString() { + return "RemoteLogConfig{" + + "remoteStorageEnable=" + remoteStorageEnable + + ", localRetentionMs=" + localRetentionMs + + ", localRetentionBytes=" + localRetentionBytes + + '}'; + } } // Visible for testing @@ -266,12 +274,7 @@ public Optional serverConfigName(String configName) { .define(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, LONG, DEFAULT_LOCAL_RETENTION_MS, atLeast(-2), MEDIUM, TopicConfig.LOCAL_LOG_RETENTION_MS_DOC) .define(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, LONG, DEFAULT_LOCAL_RETENTION_BYTES, atLeast(-2), MEDIUM, - TopicConfig.LOCAL_LOG_RETENTION_BYTES_DOC) - // 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); + TopicConfig.LOCAL_LOG_RETENTION_BYTES_DOC); } public final Set overriddenConfigs; @@ -477,15 +480,14 @@ public static void validateValues(Map props) { } /** - * Validates the configured values of the LogConfig. Should be called only by the broker. + * Validates the values of the given properties. 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) { + public static void validateBrokerLogConfigValues(Map props, + boolean isRemoteLogStorageSystemEnabled) { validateValues(props); - Boolean isRemoteLogStorageSystemEnabled = - (Boolean) props.get(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP); if (isRemoteLogStorageSystemEnabled) { validateRemoteStorageRetentionSize(props); validateRemoteStorageRetentionTime(props); @@ -494,24 +496,23 @@ public static void validateConfiguredValuesInBroker(Map props) { /** * Validates the values of the given properties. Should be called only by the broker. - * The `props` supplied should contain all the LogConfig properties and the default values should be extracted from - * the KafkaConfig. + * The `props` supplied contains the topic-level configs, + * The default values should be extracted from the KafkaConfig. * @param props The properties to be validated */ - public static void validateValuesInBroker(Map props) { + static void validateTopicLogConfigValues(Map props, + boolean isRemoteLogStorageSystemEnabled) { validateValues(props); - Boolean isRemoteLogStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); + boolean isRemoteLogStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); if (isRemoteLogStorageEnabled) { - validateRemoteStorageOnlyIfSystemEnabled(props); + validateRemoteStorageOnlyIfSystemEnabled(isRemoteLogStorageSystemEnabled); validateNoRemoteStorageForCompactedTopic(props); validateRemoteStorageRetentionSize(props); validateRemoteStorageRetentionTime(props); } } - private static void validateRemoteStorageOnlyIfSystemEnabled(Map props) { - Boolean isRemoteLogStorageSystemEnabled = - (Boolean) props.get(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP); + private static void validateRemoteStorageOnlyIfSystemEnabled(boolean isRemoteLogStorageSystemEnabled) { if (!isRemoteLogStorageSystemEnabled) { throw new ConfigException("Tiered Storage functionality is disabled in the broker. " + "Topic cannot be configured with remote log storage."); @@ -563,20 +564,21 @@ private static void validateRemoteStorageRetentionTime(Map props) { * Check that the given properties contain only valid log config names and that all values can be parsed and are valid */ public static void validate(Properties props) { - validate(props, Collections.emptyMap()); + validate(props, Collections.emptyMap(), false); } public static void validate(Properties props, - Map defaultConfigs) { + Map configuredProps, + boolean isRemoteLogStorageSystemEnabled) { validateNames(props); - if (defaultConfigs == null || defaultConfigs.isEmpty()) { + if (configuredProps == null || configuredProps.isEmpty()) { Map valueMaps = CONFIG.parse(props); validateValues(valueMaps); } else { - Map configWithBrokerDefaults = new HashMap<>(defaultConfigs); - configWithBrokerDefaults.putAll(props); - Map valueMaps = CONFIG.parse(configWithBrokerDefaults); - validateValuesInBroker(valueMaps); + Map combinedConfigs = new HashMap<>(configuredProps); + combinedConfigs.putAll(props); + Map valueMaps = CONFIG.parse(combinedConfigs); + validateTopicLogConfigValues(valueMaps, isRemoteLogStorageSystemEnabled); } } From d22d487e0449916a630856788e3793d4580d2ef6 Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Tue, 15 Aug 2023 11:53:05 +0530 Subject: [PATCH 7/9] reduce the method access specifier to private --- .../org/apache/kafka/storage/internals/log/LogConfig.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java index 609b0de961fe1..ce2880865dd38 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java @@ -500,8 +500,8 @@ public static void validateBrokerLogConfigValues(Map props, * The default values should be extracted from the KafkaConfig. * @param props The properties to be validated */ - static void validateTopicLogConfigValues(Map props, - boolean isRemoteLogStorageSystemEnabled) { + private static void validateTopicLogConfigValues(Map props, + boolean isRemoteLogStorageSystemEnabled) { validateValues(props); boolean isRemoteLogStorageEnabled = (Boolean) props.get(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG); if (isRemoteLogStorageEnabled) { From 93ac9db03c4668c5affe3609cf9a189af8533bcc Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Tue, 15 Aug 2023 16:05:19 +0530 Subject: [PATCH 8/9] Fix the unit tests. --- .../integration/kafka/admin/RemoteTopicCrudTest.scala | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala index 218f9a61af8ea..35435efe82225 100644 --- a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala +++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala @@ -23,8 +23,7 @@ import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry} import org.apache.kafka.common.TopicPartition 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.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} @@ -284,9 +283,9 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { } private def verifyRemoteLogTopicConfigs(topicConfig: Properties): Unit = { - val logOpt = brokers.head.logManager.getLog(new TopicPartition(testTopicName, 0)) - assertTrue(logOpt.isDefined) - val log = logOpt.get + val logBuffer = brokers.flatMap(_.logManager.getLog(new TopicPartition(testTopicName, 0))) + assertTrue(logBuffer.nonEmpty) + val log = logBuffer.head TestUtils.waitUntilTrue(() => { var result = true if (topicConfig.containsKey(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG)) { From 0b021785469c0e1eaa7c88962ede159dfb03376f Mon Sep 17 00:00:00 2001 From: Kamal Chandraprakash Date: Tue, 15 Aug 2023 20:50:41 +0530 Subject: [PATCH 9/9] Retry the `getLog` operation until the timeout. --- .../kafka/admin/RemoteTopicCrudTest.scala | 55 ++++++++++--------- 1 file changed, 29 insertions(+), 26 deletions(-) diff --git a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala index 35435efe82225..59adcf722bf99 100644 --- a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala +++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala @@ -283,35 +283,38 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { } private def verifyRemoteLogTopicConfigs(topicConfig: Properties): Unit = { - val logBuffer = brokers.flatMap(_.logManager.getLog(new TopicPartition(testTopicName, 0))) - assertTrue(logBuffer.nonEmpty) - val log = logBuffer.head TestUtils.waitUntilTrue(() => { - var result = true - if (topicConfig.containsKey(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG)) { - result = result && - topicConfig.getProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG).toBoolean == - log.config.remoteStorageEnable() - } - if (topicConfig.containsKey(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) { - result = result && - topicConfig.getProperty(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG).toLong == - log.config.localRetentionBytes() - } - if (topicConfig.containsKey(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG)) { - result = result && - topicConfig.getProperty(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG).toLong == log.config.localRetentionMs() - } - if (topicConfig.containsKey(TopicConfig.RETENTION_MS_CONFIG)) { - result = result && - topicConfig.getProperty(TopicConfig.RETENTION_MS_CONFIG).toLong == log.config.retentionMs - } - if (topicConfig.containsKey(TopicConfig.RETENTION_BYTES_CONFIG)) { - result = result && - topicConfig.getProperty(TopicConfig.RETENTION_BYTES_CONFIG).toLong == log.config.retentionSize + val logBuffer = brokers.flatMap(_.logManager.getLog(new TopicPartition(testTopicName, 0))) + var result = logBuffer.nonEmpty + if (result) { + if (topicConfig.containsKey(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG).toBoolean == + logBuffer.head.config.remoteStorageEnable() + } + if (topicConfig.containsKey(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG).toLong == + logBuffer.head.config.localRetentionBytes() + } + if (topicConfig.containsKey(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG).toLong == + logBuffer.head.config.localRetentionMs() + } + if (topicConfig.containsKey(TopicConfig.RETENTION_MS_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.RETENTION_MS_CONFIG).toLong == + logBuffer.head.config.retentionMs + } + if (topicConfig.containsKey(TopicConfig.RETENTION_BYTES_CONFIG)) { + result = result && + topicConfig.getProperty(TopicConfig.RETENTION_BYTES_CONFIG).toLong == + logBuffer.head.config.retentionSize + } } result - }, s"Failed to update topic config $topicConfig, actual: ${log.config}") + }, s"Failed to update topic config $topicConfig") } private def overrideProps(): Properties = {