KAFKA-7030: Add configuration to disable message down-conversion (KIP-283) by dhruvilshah3 · Pull Request #5192 · apache/kafka

dhruvilshah3 · 2018-06-11T20:27:39Z

Committer Checklist (excluded from commit message)

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

hachikuji

Thanks, left a few comments.

hachikuji · 2018-06-12T20:24:30Z

+    val fetchRequest = FetchRequest.Builder.forConsumer(Int.MaxValue, 0, createPartitionMap(1024,
+      topics)).build(1)
+    val fetchResponse = sendFetchRequest(topicMap.head._2, fetchRequest)
+    topics.foreach(tp => assertEquals(Errors.UNSUPPORTED_VERSION, fetchResponse.responseData().get(tp).error))


This assertion is satisfied even if topics doesn't contain all the expected topics. Similarly below.

hachikuji · 2018-06-12T20:26:40Z

        "exceeds this threshold. This configuration is ignored if message.timestamp.type=LogAppendTime.";
+
+    public static final String MESSAGE_DOWNCONVERSION_ENABLE_CONFIG = "message.downconversion.enable";
+    public static final String MESSAGE_DOWNCONVERSION_ENABLE_DOC = "This configuration controls whether down-conversion of message formats is enabled to satisfy consume requests.";


We should mention the impact of this config on old clients.

hachikuji · 2018-06-12T20:36:01Z

+      // For fetch requests from clients, check if down-conversion is disabled for the particular partition
+      if (!fetchRequest.isFromFollower && downConvertMagic.isDefined && !replicaManager.messageDownConversionEnabled(tp).getOrElse(true)) {
+        trace(s"Conversion to message format ${downConvertMagic.get} is disabled for partition $tp. Sending unsupported version response to $clientId.")
+        new FetchResponse.PartitionData(Errors.UNSUPPORTED_VERSION, FetchResponse.INVALID_HIGHWATERMARK,


Could we use errorResponse?

Unfortunately, no. errorResponse returns FetchResponse.PartitionData[Records] but we want FetchResponse.PartitionData[BaseRecords] here.

Found the incantation to make this work:

def errorResponse[T >: MemoryRecords <: BaseRecords](error: Errors): FetchResponse.PartitionData[T] = { new FetchResponse.PartitionData[T](error, FetchResponse.INVALID_HIGHWATERMARK, FetchResponse.INVALID_LAST_STABLE_OFFSET, FetchResponse.INVALID_LOG_START_OFFSET, null, MemoryRecords.EMPTY) }

This says that T is a supertype of MemoryRecords and a subtype of BaseRecords. Since Records and BaseRecords both satisfy this bound, it works. I already did this locally, so I'll just push to this branch and merge.

hachikuji · 2018-06-12T20:39:47Z

-        }
-      }.getOrElse(unconvertedRecords)
+      // For fetch requests from clients, check if down-conversion is disabled for the particular partition
+      if (!fetchRequest.isFromFollower && downConvertMagic.isDefined && !replicaManager.messageDownConversionEnabled(tp).getOrElse(true)) {


It's a little vexing we have to hit ReplicaManager a second time and repeat the same logic to extract the config. What if we just exposed an API to get the LogConfig directly for a partition?

hachikuji · 2018-06-12T20:49:24Z

    }

-    def convertRecords(tp: TopicPartition, unconvertedRecords: Records): BaseRecords = {
+    def convertFetchedData(tp: TopicPartition,


Perhaps we should call this maybeConvertFetchedData since the conversion may or may not happen depending on whether it is needed and permitted.

hachikuji · 2018-06-12T20:50:11Z

+      }
+    }
+
+    def convertRecords(tp: TopicPartition, unconvertedRecords: Records, downConvertMagic: Option[Byte]): BaseRecords = {


nit: This method is a little odd since the conversion only happens if downConvertMagic is not None. It seems a little more natural to move the map to the caller. Since we're not hiding much, I'd probably just suggest inlining it below.

hachikuji · 2018-06-12T21:01:47Z

+    }.toMap
+  }
+
+  private def createPartitionMap(maxPartitionBytes: Int, topicPartitions: Seq[TopicPartition],


Seems like we're borrowing some code from FetchTest. Would it make sense to extend it?

Probably wouldn't make sense to extend it. I tried extracting some logic into a utility class but looks like it needs some amount of refactoring.

hachikuji · 2018-06-12T21:02:44Z

+
+  @Test
+  def testV1FetchWithDownConversionDisabled(): Unit = {
+    initProducer()


You can do this in a @Before method.

hachikuji

Thanks, left a couple additional comments.

hachikuji · 2018-06-13T16:29:00Z

    "implement the <code>org.apache.kafka.server.policy.CreateTopicPolicy</code> interface."
  val AlterConfigPolicyClassNameDoc = "The alter configs policy class that should be used for validation. The class should " +
    "implement the <code>org.apache.kafka.server.policy.AlterConfigPolicy</code> interface."
+  val LogMessageDownConversionEnableDoc = "This configuration controls whether down-conversion of message formats is enabled to satisfy consume requests."


Seems not ideal to have this documentation in two places. I know you are just following the current pattern, but if we don't actually have any difference to communicate, perhaps we should just refer to the documentation in TopicConfig?

hachikuji · 2018-06-13T16:29:31Z

+  }
+
+  @Test
+  def testV1FetchFromReplica(): Unit = {


I think the point of this test is to ensure that the downconversion config does not apply to replicas? Probably worth spelling this out. Also, I'm wondering if we should mention this in the config documentation?

hachikuji · 2018-06-13T16:32:01Z

-      }.getOrElse(unconvertedRecords)
+      // For fetch requests from clients, check if down-conversion is disabled for the particular partition
+      if (downConvertMagic.isDefined && !fetchRequest.isFromFollower &&
+          !logConfig.map(_.messageDownConversionEnable.booleanValue()).getOrElse(true)) {


I think this can be simplified: !logConfig.forall(_.messageDownConversionEnable)?

hachikuji

LGTM, thanks for the patch!

…-283) (#5192) Add support for the topic-level `message.downconversion.enable` config as part of KIP-283.

…-283) (apache#5192) Add support for the topic-level `message.downconversion.enable` config as part of KIP-283.

Add configuration to disable message down-conversion

1622969

dhruvilshah3 force-pushed the conversion-config branch from 7ae2ec2 to 1622969 Compare June 11, 2018 20:57

dhruvilshah3 added 3 commits June 11, 2018 15:32

Add Apache license to new file

5f046da

Enable dynamic reconfiguration for message.downconversion.enable

66d41f8

Fix test

e131119

hachikuji reviewed Jun 12, 2018

View reviewed changes

Address review comments

2f98d35

dhruvilshah3 force-pushed the conversion-config branch from 2e62790 to 2f98d35 Compare June 13, 2018 00:13

hachikuji reviewed Jun 13, 2018

View reviewed changes

dhruvilshah3 and others added 2 commits June 13, 2018 14:36

Address review comments

c987d2d

Divine type incantation to make errorResponse work for BaseRecords usage

248d13e

hachikuji force-pushed the conversion-config branch from 72d8db0 to 248d13e Compare June 13, 2018 23:14

hachikuji approved these changes Jun 13, 2018

View reviewed changes

hachikuji merged commit 9a71bfb into apache:trunk Jun 14, 2018

hachikuji pushed a commit that referenced this pull request Jun 14, 2018

KAFKA-7030; Add configuration to disable message down-conversion (KIP…

0db889d

…-283) (#5192) Add support for the topic-level `message.downconversion.enable` config as part of KIP-283.

dhruvilshah3 deleted the conversion-config branch June 14, 2018 06:38

ying-zheng pushed a commit to ying-zheng/kafka that referenced this pull request Jul 6, 2018

KAFKA-7030; Add configuration to disable message down-conversion (KIP…

8888517

…-283) (apache#5192) Add support for the topic-level `message.downconversion.enable` config as part of KIP-283.

Conversation

dhruvilshah3 commented Jun 11, 2018

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Jun 13, 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!

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!

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!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

hachikuji Jun 13, 2018 •

edited

Loading