From a1ee3bf2022ef956552c8161916afebc2e2b032d Mon Sep 17 00:00:00 2001 From: huxihx Date: Thu, 31 May 2018 16:05:38 +0800 Subject: [PATCH 1/2] KAFKA-6973: TopicCommand should verify topic-level config `message.timestamp.type`. Specifying an invalid config other than `CreateTime` or `LogAppendTime` will cause broker restarting to fail. --- core/src/main/scala/kafka/log/LogConfig.scala | 2 +- .../unit/kafka/admin/TopicCommandTest.scala | 20 +++++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) 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..97689064e9903 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -225,4 +225,24 @@ 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) + + val topic = "test" + val numPartitions = 1 + + // create the topic + try { + val createOpts = new TopicCommandOptions( + Array("--partitions", numPartitions.toString, "--replication-factor", "1", "--topic", topic, + "--config", "message.timestamp.type=boom")) + TopicCommand.createTopic(zkClient, createOpts) + fail("Expected exception on invalid topic-level config.") + } catch { + case e: Exception => // topic create should fail due to the invalid config + } + } + } From 872ac967e49744569aaa2d83b4072345d7c598dd Mon Sep 17 00:00:00 2001 From: huxihx Date: Fri, 1 Jun 2018 16:18:30 +0800 Subject: [PATCH 2/2] addressed Ijuma's comments --- .../src/test/scala/unit/kafka/admin/TopicCommandTest.scala | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala index 97689064e9903..782fcf539f3f1 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -230,18 +230,15 @@ class TopicCommandTest extends ZooKeeperTestHarness with Logging with RackAwareT val brokers = List(0) TestUtils.createBrokersInZk(zkClient, brokers) - val topic = "test" - val numPartitions = 1 - // create the topic try { val createOpts = new TopicCommandOptions( - Array("--partitions", numPartitions.toString, "--replication-factor", "1", "--topic", topic, + 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 e: Exception => // topic create should fail due to the invalid config + case _: Exception => // topic creation should fail due to the invalid config } }