KAFKA-7409: Validate topic configs prior to topic creation #5651

huxihx · 2018-09-14T01:28:31Z

KAFKA-7409: Validate topic configs prior to topic creation
https://issues.apache.org/jira/browse/KAFKA-7409

Values for message.format.version and log.message.format.version should be verfied before topic creation or config change.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

https://issues.apache.org/jira/browse/KAFKA-7409 Values for `message.format.version` and `log.message.format.version` should be verfied before topic creation or config change.

hachikuji

Thanks @huxihx for picking this up so quickly. Left a few minor comments.

hachikuji · 2018-09-14T05:59:54Z

core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala

+    try {
+      val createOpts = new TopicCommandOptions(
+        Array("--partitions", "1", "--replication-factor", "1", "--topic", "test",
+          "--config", "message.message.version=boom"))


The config should be message.format.version.

hachikuji · 2018-09-14T06:03:26Z

core/src/main/scala/kafka/api/ApiVersion.scala

    KAFKA_2_1_IV1
  )

+  val allVersionsStr = (allVersions.map(_.toString) ++ allVersions.map(_.shortVersion)).toArray


Hmm.. couldn't we use something like versionMap.keys.toSeq?

core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala

omkreddy

LGTM.

One minor caveat is, when using with zk based topic creation, version info will be taken from the jar available on the client machines. The clients may be using different version of Kafka. In this case, there is a chance to create topic with invalid versions.

omkreddy · 2018-09-14T06:54:45Z

core/src/main/scala/kafka/server/AdminManager.scala

+          info(s"Error processing create topic request for topic $topic with arguments $arguments", e)
+          CreatePartitionsMetadata(topic, Map(), ApiError.fromThrowable(new InvalidConfigurationException(e.getMessage, e.getCause)))
        case e: Throwable =>
+          e.printStackTrace()


nit: unintended print?

hachikuji

Thanks for the updates. A few more small comments.

hachikuji · 2018-09-14T18:24:54Z

core/src/main/scala/kafka/server/KafkaConfig.scala

  val AutoCreateTopicsEnable = true
  val MinInSyncReplicas = 1
  val MessageDownConversionEnable = true
+  val validApiVersions = ApiVersion.allVersionsArray


Could we reference the ApiVersion.allVersionsArray directly below? This is not really a default, but a constraint.

hachikuji · 2018-09-14T18:27:49Z

core/src/main/scala/kafka/api/ApiVersion.scala

  private val versionMap: Map[String, ApiVersion] =
    allVersions.map(v => v.version -> v).toMap ++ allVersions.groupBy(_.shortVersion).map { case (k, v) => k -> v.last }

+  val allVersionsArray = versionMap.keys.seq.toArray


nit: this might be more generally useful if exposed as a Set. We can convert it to a seq or an array when we pass it to in in the config definition.

hachikuji · 2018-09-14T18:28:32Z

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

    props.put(KafkaConfig.InterBrokerProtocolVersionProp, "0.8.2.0")
    // We need to set the message format version to make the configuration valid.
-    props.put(KafkaConfig.LogMessageFormatVersionProp, "0.8.2.0")
+    props.put(KafkaConfig.LogMessageFormatVersionProp, "0.8.2")


What was the reason for changing this?

It's not a valid value actually, instead 0.8.2 is.

Haha, good point. Shall we change the InterBrokerProtocolVersion above as well?

huxihx · 2018-09-17T03:24:22Z

@hachikuji Please review again. Thanks.

ijuma · 2018-09-17T07:36:47Z

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

    assertEquals(KAFKA_0_8_2, conf2.interBrokerProtocolVersion)

-    // check that 0.8.2.0 is the same as 0.8.2.1
-    props.put(KafkaConfig.InterBrokerProtocolVersionProp, "0.8.2.1")


Is this not allowed anymore? That would be a breaking changes, right?

Yeah, @ijuma is right. We have to account for the logic inside ApiVersion.apply in order to validate the version. A custom config validator might be one option.

hachikuji · 2018-09-18T19:00:07Z

core/src/main/scala/kafka/log/LogConfig.scala

      .define(PreAllocateEnableProp, BOOLEAN, Defaults.PreAllocateEnable, MEDIUM, PreAllocateEnableDoc,
        KafkaConfig.LogPreAllocateProp)
-      .define(MessageFormatVersionProp, STRING, Defaults.MessageFormatVersion, in(ApiVersion.allValidVersions.toArray:_*), MEDIUM, MessageFormatVersionDoc,
+      .define(MessageFormatVersionProp, STRING, Defaults.MessageFormatVersion, ValidPrefixString.in(ApiVersion.allValidVersions.toArray:_*), MEDIUM, MessageFormatVersionDoc,


Since we only have on use case for this at the moment for validating ApiVersion, I wonder if it would be simpler to have a custom ApiVersion.Validator which invokes ApiVersion.apply directly? My concern is that we have two separate paths for validation. The logic seems consistent as far as I can tell, but we may change the validation logic in the future.

Correct me if I am wrong. Currently, Validator and its implementations are in clients codebase which does not see any classes from within the core. It cannot invoke ApiVersion.apply. Right?

To clarify, I am suggesting that we add a custom Validator implementation inside ApiVersion. We need not put it in the clients module since ApiVersion is in core.

huxihx · 2018-09-26T01:38:17Z

@hachikuji Please review again. Thanks.

hachikuji

LGTM. Thanks @huxihx!

… altering configs (apache#5651) Values for `message.format.version` and `log.message.format.version` should be verified before topic creation or config change. Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

huxihx added 2 commits September 14, 2018 09:27

KAFKA-7409: Validate topic configs prior to topic creation

8d122d7

https://issues.apache.org/jira/browse/KAFKA-7409 Values for `message.format.version` and `log.message.format.version` should be verfied before topic creation or config change.

Make short versions of ApiVersion as valid values

01238b1

hachikuji self-assigned this Sep 14, 2018

hachikuji reviewed Sep 14, 2018

View reviewed changes

addressed Jason's comments

e4090aa

omkreddy reviewed Sep 14, 2018

View reviewed changes

removed unintended print

a03246d

hachikuji reviewed Sep 14, 2018

View reviewed changes

addressed Json's comments

6860b6c

ijuma reviewed Sep 17, 2018

View reviewed changes

Added a custom validator

c3e9183

hachikuji reviewed Sep 18, 2018

View reviewed changes

Implemented custom Validator in ApiVersion

d1530ed

hachikuji approved these changes Sep 28, 2018

View reviewed changes

hachikuji merged commit 70d90c3 into apache:trunk Sep 28, 2018

KAFKA-7409: Validate topic configs prior to topic creation #5651

KAFKA-7409: Validate topic configs prior to topic creation #5651

Uh oh!

Conversation

huxihx commented Sep 14, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

omkreddy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Sep 15, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

huxihx commented Sep 17, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

huxihx commented Sep 26, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

huxihx commented Sep 14, 2018 •

edited

Loading

hachikuji Sep 15, 2018 •

edited

Loading

huxihx commented Sep 17, 2018 •

edited

Loading