kamalcph commented on code in PR #14161:
URL: https://github.com/apache/kafka/pull/14161#discussion_r1304548979
##########
core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala:
##########
@@ -301,6 +300,70 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
// "Remote log segments should be deleted only once by the leader")
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+ @ValueSource(strings = Array("zk"))
+ def
testClusterWideDisablementOfTieredStorageWithEnabledTieredTopicZK(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))
+
+ assertThrows(classOf[ConfigException], () => recreateBrokers(startup =
true))
+ }
Review Comment:
```suggestion
@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")) {
assertThrows(classOf[ConfigException], () => recreateBrokers(startup =
true))
} else {
recreateBrokers(startup = true)
assertTrue(faultHandler.firstException().getCause.isInstanceOf[ConfigException])
// Normally the exception is thrown as part of the TearDown method of
the parent class(es). We would like to not do this.
faultHandler.setIgnore(true)
}
}
```
With this, we can have a common test for both the modes.
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -870,12 +870,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:
Can we allow the tiered storage functionality disablement since no data is
in remote?
##########
core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala:
##########
@@ -301,6 +300,70 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
// "Remote log segments should be deleted only once by the leader")
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+ @ValueSource(strings = Array("zk"))
+ def
testClusterWideDisablementOfTieredStorageWithEnabledTieredTopicZK(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))
+
+ assertThrows(classOf[ConfigException], () => recreateBrokers(startup =
true))
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+ @ValueSource(strings = Array("zk"))
+ def
testClusterWithoutTieredStorageStartsSuccessfullyIfTopicWithTieringDisabledZK(quorum:
String): Unit = {
Review Comment:
```suggestion
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
@ValueSource(strings = Array("zk", "kraft"))
def
testClusterWithoutTieredStorageStartsSuccessfullyIfTopicWithTieringDisabled(quorum:
String): Unit = {
```
With this, we can have a common test for both the modes.
--
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
