diff --git a/core/src/main/scala/kafka/log/LogConfig.scala b/core/src/main/scala/kafka/log/LogConfig.scala index 0db49e76e2e7d..158209a1fc0a2 100755 --- a/core/src/main/scala/kafka/log/LogConfig.scala +++ b/core/src/main/scala/kafka/log/LogConfig.scala @@ -254,7 +254,7 @@ object LogConfig { KafkaConfig.LogPreAllocateProp) .define(MessageFormatVersionProp, STRING, Defaults.MessageFormatVersion, MEDIUM, MessageFormatVersionDoc, KafkaConfig.LogMessageFormatVersionProp) - .define(MessageTimestampTypeProp, STRING, Defaults.MessageTimestampType, MEDIUM, MessageTimestampTypeDoc, + .define(MessageTimestampTypeProp, STRING, Defaults.MessageTimestampType, in("CreateTime", "LogAppendTime"), MEDIUM, MessageTimestampTypeDoc, KafkaConfig.LogMessageTimestampTypeProp) .define(MessageTimestampDifferenceMaxMsProp, LONG, Defaults.MessageTimestampDifferenceMaxMs, atLeast(0), MEDIUM, MessageTimestampDifferenceMaxMsDoc, KafkaConfig.LogMessageTimestampDifferenceMaxMsProp) diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala index 6a276df1e7c00..782fcf539f3f1 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -225,4 +225,21 @@ class TopicCommandTest extends ZooKeeperTestHarness with Logging with RackAwareT assertTrue(output.contains(topic) && output.contains(markedForDeletionList)) } + @Test + def testInvalidTopicLevelConfig(): Unit = { + val brokers = List(0) + TestUtils.createBrokersInZk(zkClient, brokers) + + // create the topic + try { + val createOpts = new TopicCommandOptions( + Array("--partitions", "1", "--replication-factor", "1", "--topic", "test", + "--config", "message.timestamp.type=boom")) + TopicCommand.createTopic(zkClient, createOpts) + fail("Expected exception on invalid topic-level config.") + } catch { + case _: Exception => // topic creation should fail due to the invalid config + } + } + }