divijvaidya commented on code in PR #14161: URL: https://github.com/apache/kafka/pull/14161#discussion_r1308425188
########## core/src/main/scala/kafka/log/LogManager.scala: ########## @@ -871,12 +871,17 @@ class LogManager(logDirs: Seq[File], * Update the configuration of the provided topic. */ def updateTopicConfig(topic: String, - newTopicConfig: Properties): Unit = { + newTopicConfig: Properties, + isRemoteLogStorageSystemEnabled: Boolean): Unit = { topicConfigUpdated(topic) val logs = logsByTopic(topic) + // Combine the default properties with the overrides in zk to create the new LogConfig + val newLogConfig = LogConfig.fromProps(currentDefaultConfig.originals, newTopicConfig) + // We would like to validate the configuration no matter whether the logs have materialised on disk or not. + // Otherwise we risk someone creating a tiered-topic, disabling Tiered Storage cluster-wide and the check + // failing since the logs for the topic are non-existent. + LogConfig.validateRemoteStorageOnlyIfSystemEnabled(newLogConfig.values(), isRemoteLogStorageSystemEnabled, true) Review Comment: nit it's beneficial to have named arguments for readibility. See: > Use named arguments when passing in literal values if the meaning is at all unclear, for example instead of Utils.delete(true) prefer Utils.delete(recursive=true). from https://kafka.apache.org/coding-guide ########## core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala: ########## @@ -301,6 +300,43 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { // "Remote log segments should be deleted only once by the leader") } + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk", "kraft")) + def testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic(quorum: String): Unit = { + val topicConfig = new Properties() + topicConfig.setProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true") + + TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, numPartitions, brokerCount, + topicConfig = topicConfig) + + val tsDisabledProps = TestUtils.createBrokerConfigs(1, zkConnectOrNull).head + instanceConfigs = List(KafkaConfig.fromProps(tsDisabledProps)) + + if (quorum.equals("zk")) { Review Comment: nit You can also use `isKRaftTest()` here and move Zk scenario to else. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org