KAFKA-6238; Fix inter-broker protocol message format compatibility check #4583

hachikuji · 2018-02-17T08:05:27Z

This patch fixes a bug in the validation of the inter-broker protocol and the message format version. We should allow the configured message format api version to be greater than the inter-broker protocol api version as long as the actual message format versions are equal. For example, if the message format version is set to 1.0, it is fine for the inter-broker protocol version to be 0.11.0 because they both use message format v2.

I have added a unit test which checks compatibility for all combinations of the message format version and the inter-broker protocol version.

Committer Checklist (excluded from commit message)

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

tedyu · 2018-02-17T15:43:38Z

lgtm

Please correct the JIRA number in the subject

ijuma · 2018-02-20T19:38:35Z

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

+      case 1 => "0.10.0"
+      case 2 => "0.11.0"
+      case _ => throw new IllegalArgumentException(s"Invalid message format version $messageFormatVersion")
+    }


Why not figure out these numbers automatically?

I was trying to avoid the internal version numbers.

ijuma · 2018-02-20T19:44:41Z

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

+    intercept[IllegalArgumentException] {
+      KafkaConfig.fromProps(props)
+    }
+  }


Nit: I wonder if these 2 methods should be inner defs. You could also potentially have another inner def like:

def kafkaConfig(interBrokerProtocol: ApiVersion, messageFormat: ApiVersion): KafkaConfig = { val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181) props.put(KafkaConfig.InterBrokerProtocolVersionProp, interBrokerProtocol.version) props.put(KafkaConfig.LogMessageFormatVersionProp, messageFormat.version) KafkaConfig.fromProps(props) }

ijuma

Thanks for the update, LGTM.

hachikuji · 2018-02-21T06:28:51Z

@ijuma Thanks for the review. I've renamed the enum to the more concise RecordFormat to be consistent with the naming in the rest of the package. Does that sound reasonable?

This patch fixes a bug in the validation of the inter-broker protocol and the message format version. We should allow the configured message format api version to be greater than the inter-broker protocol api version as long as the actual message format versions are equal. For example, if the message format version is set to 1.0, it is fine for the inter-broker protocol version to be 0.11.0 because they both use message format v2. I have added a unit test which checks compatibility for all combinations of the message format version and the inter-broker protocol version. Author: Jason Gustafson <jason@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk> Closes #4583 from hachikuji/KAFKA-6328-REOPENED (cherry picked from commit 660c0c0) Signed-off-by: Damian Guy <damian.guy@gmail.com>

dguy · 2018-02-21T09:45:10Z

Merged to trunk and 1.1

ijuma

Thanks. Left some minor comments.

Also worth filing a JIRA for making terminology consistent across the board for magic, message format version, record format/version. Or maybe there is a JIRA for that already.

ijuma · 2018-02-21T06:57:24Z

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

+  }
+
+  def minVersionForMessageFormat(messageFormatVersion: RecordFormat): String = {
+    messageFormatVersion match {


Replace all messageFormatVersion with recordFormat?

ijuma · 2018-02-21T06:58:40Z

clients/src/main/java/org/apache/kafka/common/record/RecordFormat.java

+ */
+package org.apache.kafka.common.record;
+
+public enum RecordFormat {


I was wondering between RecordFormat and RecordVersion. I assume you think the former is clearer?

Also, maybe we want to add brief javadoc explaining what this represents (particularly given the fact that configs still talk about message format version).

Let's compromise and go with RecordMagic 😜

I think I do prefer RecordVersion after thinking about it. The current name suggests that it has something to do with the schema.

ijuma · 2018-02-21T07:00:17Z

clients/src/main/java/org/apache/kafka/common/record/RecordFormat.java

+        this.value = (byte) value;
+    }
+
+    public static RecordFormat lookup(byte version) {


Should this parameter name be consistent with the field name?

ijuma · 2018-02-21T07:02:49Z

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

@@ -1002,7 +1002,7 @@ class ReplicaManager(val config: KafkaConfig,
  }

  def getMagic(topicPartition: TopicPartition): Option[Byte] =
-    getReplica(topicPartition).flatMap(_.log.map(_.config.messageFormatVersion.messageFormatVersion))
+    getReplica(topicPartition).flatMap(_.log.map(_.config.messageFormatVersion.messageFormatVersion.value))


It's kind of weird that we have the double messageFormatVersion, but I guess that's unavoidable given the (IMHO mistaken) choice to couple message format versions with api versions.

Yes, it would have been simpler and bugs like this wouldn't have been possible if we exposed the message format version directly in the config.

ijuma · 2018-02-21T16:42:21Z

clients/src/main/java/org/apache/kafka/common/record/RecordFormat.java

+            case 1: return V1;
+            case 2: return V2;
+            default: throw new IllegalArgumentException("Unknown format version: " + version);
+        }


You could just do values()[version] after the appropriate bounds check at the start of the method. So that this doesn't have to change as we add versions. One downside is that values causes a copy. So we could save a copy as a private field instead (this is sadly a common pattern with enums).

hachikuji · 2018-02-21T17:24:58Z

@ijuma Thanks for the comments. I'll submit a minor follow-up.

This patch fixes a bug in the validation of the inter-broker protocol and the message format version. We should allow the configured message format api version to be greater than the inter-broker protocol api version as long as the actual message format versions are equal. For example, if the message format version is set to 1.0, it is fine for the inter-broker protocol version to be 0.11.0 because they both use message format v2. I have added a unit test which checks compatibility for all combinations of the message format version and the inter-broker protocol version. Author: Jason Gustafson <jason@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk> Closes #4583 from hachikuji/KAFKA-6328-REOPENED

Also include a few clean-ups: * Method/variable/parameter renames to make them consistent with the class name * Return `ApiVersion` from `minSupportedFor` * Use `values` to remove some code duplication * Reduce duplication in `ApiVersion` by introducing the `shortVersion` method and building the versions map programatically * Avoid unnecessary `regex` in `ApiVersion.apply` * Added scaladoc to a few methods Some of these were originally discussed in: #4583 (review) Added a test for `ApiVersion.shortVersion`. Relying on existing tests for the rest since there is no change in behaviour. Reviewers: Jason Gustafson <jason@confluent.io>

Also include a few clean-ups: * Method/variable/parameter renames to make them consistent with the class name * Return `ApiVersion` from `minSupportedFor` * Use `values` to remove some code duplication * Reduce duplication in `ApiVersion` by introducing the `shortVersion` method and building the versions map programatically * Avoid unnecessary `regex` in `ApiVersion.apply` * Added scaladoc to a few methods Some of these were originally discussed in: apache#4583 (review) Added a test for `ApiVersion.shortVersion`. Relying on existing tests for the rest since there is no change in behaviour. Reviewers: Jason Gustafson <jason@confluent.io>

KAFKA-6328; Fix inter-broker protocol message format compatibility check

cfc8d32

hachikuji changed the title ~~KAFKA-6328; Fix inter-broker protocol message format compatibility check~~ KAFKA-6238; Fix inter-broker protocol message format compatibility check Feb 18, 2018

Update upgrade notes for 1.1

9a044fc

ijuma reviewed Feb 20, 2018

View reviewed changes

Jason Gustafson added 2 commits February 20, 2018 12:53

Refactor test case to use inner def

16c61ef

Add enum for message format version and use it in ApiVersion

4cd1a13

ijuma approved these changes Feb 21, 2018

View reviewed changes

Change MessageFormatVersion to RecordFormat

7dd241c

Minor naming tweaks

c138559

dguy closed this in 660c0c0 Feb 21, 2018

ijuma reviewed Feb 21, 2018

View reviewed changes

ijuma mentioned this pull request Apr 2, 2018

MINOR: Rename RecordFormat to RecordVersion #4809

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6238; Fix inter-broker protocol message format compatibility check #4583

KAFKA-6238; Fix inter-broker protocol message format compatibility check #4583

hachikuji commented Feb 17, 2018 •

edited

Loading

tedyu commented Feb 17, 2018

ijuma Feb 20, 2018 •

edited

Loading

hachikuji Feb 20, 2018

ijuma Feb 20, 2018

ijuma left a comment

hachikuji commented Feb 21, 2018

dguy commented Feb 21, 2018

ijuma left a comment

ijuma Feb 21, 2018

ijuma Feb 21, 2018

hachikuji Feb 21, 2018

ijuma Feb 21, 2018

ijuma Feb 21, 2018

hachikuji Feb 21, 2018

ijuma Feb 21, 2018

hachikuji commented Feb 21, 2018

KAFKA-6238; Fix inter-broker protocol message format compatibility check #4583

KAFKA-6238; Fix inter-broker protocol message format compatibility check #4583

Conversation

hachikuji commented Feb 17, 2018 • edited Loading

Committer Checklist (excluded from commit message)

tedyu commented Feb 17, 2018

ijuma Feb 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

hachikuji commented Feb 21, 2018

dguy commented Feb 21, 2018

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji commented Feb 21, 2018

hachikuji commented Feb 17, 2018 •

edited

Loading

ijuma Feb 20, 2018 •

edited

Loading