divijvaidya commented on code in PR #14161:
URL: https://github.com/apache/kafka/pull/14161#discussion_r1294991302
##########
core/src/test/scala/unit/kafka/server/KafkaServerTest.scala:
##########
@@ -154,6 +155,96 @@ class KafkaServerTest extends QuorumTestHarness {
server.shutdown()
}
+ @Test
+ def testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic(): Unit
= {
+ val tsEnabledProps = TestUtils.createBrokerConfigs(1, zkConnect).head
+
tsEnabledProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP,
true.toString)
+
tsEnabledProps.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
+
"org.apache.kafka.server.log.remote.storage.NoOpRemoteLogMetadataManager")
+
tsEnabledProps.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP,
+ "org.apache.kafka.server.log.remote.storage.NoOpRemoteStorageManager")
+
+ val server = TestUtils.createServer(KafkaConfig.fromProps(tsEnabledProps))
+ server.remoteLogManagerOpt match {
+ case Some(_) =>
+ case None => fail("RemoteLogManager should be initialized")
+ }
+
+ val topicProps = new Properties()
+ topicProps.setProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG,
true.toString)
+
+ TestUtils.createTopic(zkClient = server.zkClient, topic = "batman",
servers = Seq(server), topicConfig = topicProps)
+
+ server.shutdown()
+
+ val tsDisabledProps = TestUtils.createBrokerConfigs(1, zkConnect).head
+
+ assertThrows(classOf[ConfigException], () =>
TestUtils.createServer(KafkaConfig.fromProps(tsDisabledProps)))
+ }
+
+ @Test
+ def testClusterWideDisablementOfTieredStorageWithDisabledTieredTopic(): Unit
= {
+ val tsEnabledProps = TestUtils.createBrokerConfigs(1, zkConnect).head
+
tsEnabledProps.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP,
true.toString)
+
tsEnabledProps.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
+
"org.apache.kafka.server.log.remote.storage.NoOpRemoteLogMetadataManager")
+
tsEnabledProps.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP,
+ "org.apache.kafka.server.log.remote.storage.NoOpRemoteStorageManager")
+
+ var server = TestUtils.createServer(KafkaConfig.fromProps(tsEnabledProps))
+ server.remoteLogManagerOpt match {
+ case Some(_) =>
+ case None => fail("RemoteLogManager should be initialized")
+ }
+
+ val topicProps = new Properties()
+ topicProps.setProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG,
false.toString)
+
+ TestUtils.createTopic(zkClient = server.zkClient, topic = "batman",
servers = Seq(server), topicConfig = topicProps)
+
+ server.shutdown()
+
+ val tsDisabledProps = TestUtils.createBrokerConfigs(1, zkConnect).head
+
+ server = TestUtils.createServer(KafkaConfig.fromProps(tsDisabledProps))
+
+ server.shutdown()
+ }
+
+ @Test
+ def
testClusterWithoutTieredStorageFailsOnStartupIfTopicWithTieringEnabled(): Unit
= {
+ val serverProps = TestUtils.createBrokerConfigs(1, zkConnect).head
+
+ val server = TestUtils.createServer(KafkaConfig.fromProps(serverProps))
+
+ val topicProps = new Properties()
+ topicProps.setProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG,
true.toString)
+
+ TestUtils.createTopic(zkClient = server.zkClient, topic = "batman",
servers = Seq(server), topicConfig = topicProps)
+
+ server.shutdown()
Review Comment:
this should be done in finally again so that if an exception is thrown, we
are correctly cleaning up resources used by test i.e. closing the server.
--
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
